http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/rdkafka_performance.c ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.4/examples/rdkafka_performance.c b/thirdparty/librdkafka-0.11.4/examples/rdkafka_performance.c new file mode 100644 index 0000000..bceb23b --- /dev/null +++ b/thirdparty/librdkafka-0.11.4/examples/rdkafka_performance.c @@ -0,0 +1,1651 @@ +/* + * librdkafka - Apache Kafka C library + * + * Copyright (c) 2012, 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. + */ + +/** + * Apache Kafka consumer & producer performance tester + * using the Kafka driver from librdkafka + * (https://github.com/edenhill/librdkafka) + */ + +#ifdef _MSC_VER +#define _CRT_SECURE_NO_WARNINGS /* Silence nonsense on MSVC */ +#endif + +#include "../src/rd.h" + +#define _GNU_SOURCE /* for strndup() */ +#include <ctype.h> +#include <signal.h> +#include <string.h> +#include <errno.h> + +/* Typical include path would be <librdkafka/rdkafka.h>, but this program + * is built from within the librdkafka source tree and thus differs. */ +#include "rdkafka.h" /* for Kafka driver */ +/* Do not include these defines from your program, they will not be + * provided by librdkafka. */ +#include "rd.h" +#include "rdtime.h" + +#ifdef _MSC_VER +#include "../win32/wingetopt.h" +#include "../win32/wintime.h" +#endif + + +static int run = 1; +static int forever = 1; +static rd_ts_t dispintvl = 1000; +static int do_seq = 0; +static int exit_after = 0; +static int exit_eof = 0; +static FILE *stats_fp; +static int dr_disp_div; +static int verbosity = 1; +static int latency_mode = 0; +static int report_offset = 0; +static FILE *latency_fp = NULL; +static int msgcnt = -1; +static int incremental_mode = 0; +static int partition_cnt = 0; +static int eof_cnt = 0; +static int with_dr = 1; +static int read_hdrs = 0; + + +static void stop (int sig) { + if (!run) + exit(0); + run = 0; +} + +static long int msgs_wait_cnt = 0; +static long int msgs_wait_produce_cnt = 0; +static rd_ts_t t_end; +static rd_kafka_t *global_rk; + +struct avg { + int64_t val; + int cnt; + uint64_t ts_start; +}; + +static struct { + rd_ts_t t_start; + rd_ts_t t_end; + rd_ts_t t_end_send; + uint64_t msgs; + uint64_t msgs_last; + uint64_t msgs_dr_ok; + uint64_t msgs_dr_err; + uint64_t bytes_dr_ok; + uint64_t bytes; + uint64_t bytes_last; + uint64_t tx; + uint64_t tx_err; + uint64_t avg_rtt; + uint64_t offset; + rd_ts_t t_fetch_latency; + rd_ts_t t_last; + rd_ts_t t_enobufs_last; + rd_ts_t t_total; + rd_ts_t latency_last; + rd_ts_t latency_lo; + rd_ts_t latency_hi; + rd_ts_t latency_sum; + int latency_cnt; + int64_t last_offset; +} cnt; + + +uint64_t wall_clock (void) { + struct timeval tv; + gettimeofday(&tv, NULL); + return ((uint64_t)tv.tv_sec * 1000000LLU) + + ((uint64_t)tv.tv_usec); +} + +static void err_cb (rd_kafka_t *rk, int err, const char *reason, void *opaque) { + printf("%% ERROR CALLBACK: %s: %s: %s\n", + rd_kafka_name(rk), rd_kafka_err2str(err), reason); +} + +static void throttle_cb (rd_kafka_t *rk, const char *broker_name, + int32_t broker_id, int throttle_time_ms, + void *opaque) { + printf("%% THROTTLED %dms by %s (%"PRId32")\n", throttle_time_ms, + broker_name, broker_id); +} + +static void offset_commit_cb (rd_kafka_t *rk, rd_kafka_resp_err_t err, + rd_kafka_topic_partition_list_t *offsets, + void *opaque) { + int i; + + if (err || verbosity >= 2) + printf("%% Offset commit of %d partition(s): %s\n", + offsets->cnt, rd_kafka_err2str(err)); + + for (i = 0 ; i < offsets->cnt ; i++) { + rd_kafka_topic_partition_t *rktpar = &offsets->elems[i]; + if (rktpar->err || verbosity >= 2) + printf("%% %s [%"PRId32"] @ %"PRId64": %s\n", + rktpar->topic, rktpar->partition, + rktpar->offset, rd_kafka_err2str(err)); + } +} + +/** + * @brief Add latency measurement + */ +static void latency_add (int64_t ts, const char *who) { + if (ts > cnt.latency_hi) + cnt.latency_hi = ts; + if (!cnt.latency_lo || ts < cnt.latency_lo) + cnt.latency_lo = ts; + cnt.latency_last = ts; + cnt.latency_cnt++; + cnt.latency_sum += ts; + if (latency_fp) + fprintf(latency_fp, "%"PRIu64"\n", ts); +} + + +static void msg_delivered (rd_kafka_t *rk, + const rd_kafka_message_t *rkmessage, void *opaque) { + static rd_ts_t last; + rd_ts_t now = rd_clock(); + static int msgs; + + msgs++; + + msgs_wait_cnt--; + + if (rkmessage->err) + cnt.msgs_dr_err++; + else { + cnt.msgs_dr_ok++; + cnt.bytes_dr_ok += rkmessage->len; + } + + if (latency_mode) { + /* Extract latency */ + int64_t source_ts; + if (sscanf(rkmessage->payload, "LATENCY:%"SCNd64, + &source_ts) == 1) + latency_add(wall_clock() - source_ts, "producer"); + } + + + if ((rkmessage->err && + (cnt.msgs_dr_err < 50 || + !(cnt.msgs_dr_err % (dispintvl / 1000)))) || + !last || msgs_wait_cnt < 5 || + !(msgs_wait_cnt % dr_disp_div) || + (now - last) >= dispintvl * 1000 || + verbosity >= 3) { + if (rkmessage->err && verbosity >= 2) + printf("%% Message delivery failed: %s [%"PRId32"]: " + "%s (%li remain)\n", + rd_kafka_topic_name(rkmessage->rkt), + rkmessage->partition, + rd_kafka_err2str(rkmessage->err), + msgs_wait_cnt); + else if (verbosity > 2) + printf("%% Message delivered (offset %"PRId64"): " + "%li remain\n", + rkmessage->offset, msgs_wait_cnt); + if (verbosity >= 3 && do_seq) + printf(" --> \"%.*s\"\n", + (int)rkmessage->len, + (const char *)rkmessage->payload); + last = now; + } + + if (report_offset) + cnt.last_offset = rkmessage->offset; + + if (msgs_wait_produce_cnt == 0 && msgs_wait_cnt == 0 && !forever) { + if (verbosity >= 2) + printf("All messages delivered!\n"); + t_end = rd_clock(); + run = 0; + } + + if (exit_after && exit_after <= msgs) { + printf("%% Hard exit after %i messages, as requested\n", + exit_after); + exit(0); + } +} + + +static void msg_consume (rd_kafka_message_t *rkmessage, void *opaque) { + + if (rkmessage->err) { + if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { + cnt.offset = rkmessage->offset; + + if (verbosity >= 1) + printf("%% Consumer reached end of " + "%s [%"PRId32"] " + "message queue at offset %"PRId64"\n", + rd_kafka_topic_name(rkmessage->rkt), + rkmessage->partition, rkmessage->offset); + + if (exit_eof && ++eof_cnt == partition_cnt) + run = 0; + + return; + } + + printf("%% Consume error for topic \"%s\" [%"PRId32"] " + "offset %"PRId64": %s\n", + rkmessage->rkt ? rd_kafka_topic_name(rkmessage->rkt):"", + rkmessage->partition, + rkmessage->offset, + rd_kafka_message_errstr(rkmessage)); + + if (rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION || + rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC) + run = 0; + + cnt.msgs_dr_err++; + return; + } + + /* Start measuring from first message received */ + if (!cnt.t_start) + cnt.t_start = cnt.t_last = rd_clock(); + + cnt.offset = rkmessage->offset; + cnt.msgs++; + cnt.bytes += rkmessage->len; + + if (verbosity >= 3 || + (verbosity >= 2 && !(cnt.msgs % 1000000))) + printf("@%"PRId64": %.*s: %.*s\n", + rkmessage->offset, + (int)rkmessage->key_len, (char *)rkmessage->key, + (int)rkmessage->len, (char *)rkmessage->payload); + + + if (latency_mode) { + int64_t remote_ts, ts; + + if (rkmessage->len > 8 && + !memcmp(rkmessage->payload, "LATENCY:", 8) && + sscanf(rkmessage->payload, "LATENCY:%"SCNd64, + &remote_ts) == 1) { + ts = wall_clock() - remote_ts; + if (ts > 0 && ts < (1000000 * 60 * 5)) { + latency_add(ts, "consumer"); + } else { + if (verbosity >= 1) + printf("Received latency timestamp is too far off: %"PRId64"us (message offset %"PRId64"): ignored\n", + ts, rkmessage->offset); + } + } else if (verbosity > 1) + printf("not a LATENCY payload: %.*s\n", + (int)rkmessage->len, + (char *)rkmessage->payload); + + } + + if (read_hdrs) { + rd_kafka_headers_t *hdrs; + /* Force parsing of headers but don't do anything with them. */ + rd_kafka_message_headers(rkmessage, &hdrs); + } + + if (msgcnt != -1 && (int)cnt.msgs >= msgcnt) + run = 0; +} + + +static void rebalance_cb (rd_kafka_t *rk, + rd_kafka_resp_err_t err, + rd_kafka_topic_partition_list_t *partitions, + void *opaque) { + + switch (err) + { + case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS: + fprintf(stderr, + "%% Group rebalanced: %d partition(s) assigned\n", + partitions->cnt); + eof_cnt = 0; + partition_cnt = partitions->cnt; + rd_kafka_assign(rk, partitions); + break; + + case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS: + fprintf(stderr, + "%% Group rebalanced: %d partition(s) revoked\n", + partitions->cnt); + eof_cnt = 0; + partition_cnt = 0; + rd_kafka_assign(rk, NULL); + break; + + default: + break; + } +} + + +/** + * Find and extract single value from a two-level search. + * First find 'field1', then find 'field2' and extract its value. + * Returns 0 on miss else the value. + */ +static uint64_t json_parse_fields (const char *json, const char **end, + const char *field1, const char *field2) { + const char *t = json; + const char *t2; + int len1 = (int)strlen(field1); + int len2 = (int)strlen(field2); + + while ((t2 = strstr(t, field1))) { + uint64_t v; + + t = t2; + t += len1; + + /* Find field */ + if (!(t2 = strstr(t, field2))) + continue; + t2 += len2; + + while (isspace((int)*t2)) + t2++; + + v = strtoull(t2, (char **)&t, 10); + if (t2 == t) + continue; + + *end = t; + return v; + } + + *end = t + strlen(t); + return 0; +} + +/** + * Parse various values from rdkafka stats + */ +static void json_parse_stats (const char *json) { + const char *t; +#define MAX_AVGS 100 /* max number of brokers to scan for rtt */ + uint64_t avg_rtt[MAX_AVGS+1]; + int avg_rtt_i = 0; + + /* Store totals at end of array */ + avg_rtt[MAX_AVGS] = 0; + + /* Extract all broker RTTs */ + t = json; + while (avg_rtt_i < MAX_AVGS && *t) { + avg_rtt[avg_rtt_i] = json_parse_fields(t, &t, + "\"rtt\":", + "\"avg\":"); + + /* Skip low RTT values, means no messages are passing */ + if (avg_rtt[avg_rtt_i] < 100 /*0.1ms*/) + continue; + + + avg_rtt[MAX_AVGS] += avg_rtt[avg_rtt_i]; + avg_rtt_i++; + } + + if (avg_rtt_i > 0) + avg_rtt[MAX_AVGS] /= avg_rtt_i; + + cnt.avg_rtt = avg_rtt[MAX_AVGS]; +} + + +static int stats_cb (rd_kafka_t *rk, char *json, size_t json_len, + void *opaque) { + + /* Extract values for our own stats */ + json_parse_stats(json); + + if (stats_fp) + fprintf(stats_fp, "%s\n", json); + return 0; +} + +#define _OTYPE_TAB 0x1 /* tabular format */ +#define _OTYPE_SUMMARY 0x2 /* summary format */ +#define _OTYPE_FORCE 0x4 /* force output regardless of interval timing */ +static void print_stats (rd_kafka_t *rk, + int mode, int otype, const char *compression) { + rd_ts_t now = rd_clock(); + rd_ts_t t_total; + static int rows_written = 0; + int print_header; + double latency_avg = 0.0f; + char extra[512]; + int extra_of = 0; + *extra = '\0'; + + if (!(otype & _OTYPE_FORCE) && + (((otype & _OTYPE_SUMMARY) && verbosity == 0) || + cnt.t_last + dispintvl > now)) + return; + + print_header = !rows_written ||(verbosity > 0 && !(rows_written % 20)); + + if (cnt.t_end_send) + t_total = cnt.t_end_send - cnt.t_start; + else if (cnt.t_end) + t_total = cnt.t_end - cnt.t_start; + else if (cnt.t_start) + t_total = now - cnt.t_start; + else + t_total = 1; + + if (latency_mode && cnt.latency_cnt) + latency_avg = (double)cnt.latency_sum / + (double)cnt.latency_cnt; + + if (mode == 'P') { + + if (otype & _OTYPE_TAB) { +#define ROW_START() do {} while (0) +#define COL_HDR(NAME) printf("| %10.10s ", (NAME)) +#define COL_PR64(NAME,VAL) printf("| %10"PRIu64" ", (VAL)) +#define COL_PRF(NAME,VAL) printf("| %10.2f ", (VAL)) +#define ROW_END() do { \ + printf("\n"); \ + rows_written++; \ + } while (0) + + if (print_header) { + /* First time, print header */ + ROW_START(); + COL_HDR("elapsed"); + COL_HDR("msgs"); + COL_HDR("bytes"); + COL_HDR("rtt"); + COL_HDR("dr"); + COL_HDR("dr_m/s"); + COL_HDR("dr_MB/s"); + COL_HDR("dr_err"); + COL_HDR("tx_err"); + COL_HDR("outq"); + if (report_offset) + COL_HDR("offset"); + if (latency_mode) { + COL_HDR("lat_curr"); + COL_HDR("lat_avg"); + COL_HDR("lat_lo"); + COL_HDR("lat_hi"); + } + + ROW_END(); + } + + ROW_START(); + COL_PR64("elapsed", t_total / 1000); + COL_PR64("msgs", cnt.msgs); + COL_PR64("bytes", cnt.bytes); + COL_PR64("rtt", cnt.avg_rtt / 1000); + COL_PR64("dr", cnt.msgs_dr_ok); + COL_PR64("dr_m/s", + ((cnt.msgs_dr_ok * 1000000) / t_total)); + COL_PRF("dr_MB/s", + (float)((cnt.bytes_dr_ok) / (float)t_total)); + COL_PR64("dr_err", cnt.msgs_dr_err); + COL_PR64("tx_err", cnt.tx_err); + COL_PR64("outq", + rk ? (uint64_t)rd_kafka_outq_len(rk) : 0); + if (report_offset) + COL_PR64("offset", (uint64_t)cnt.last_offset); + if (latency_mode) { + COL_PRF("lat_curr", cnt.latency_last / 1000.0f); + COL_PRF("lat_avg", latency_avg / 1000.0f); + COL_PRF("lat_lo", cnt.latency_lo / 1000.0f); + COL_PRF("lat_hi", cnt.latency_hi / 1000.0f); + } + ROW_END(); + } + + if (otype & _OTYPE_SUMMARY) { + printf("%% %"PRIu64" messages produced " + "(%"PRIu64" bytes), " + "%"PRIu64" delivered " + "(offset %"PRId64", %"PRIu64" failed) " + "in %"PRIu64"ms: %"PRIu64" msgs/s and " + "%.02f MB/s, " + "%"PRIu64" produce failures, %i in queue, " + "%s compression\n", + cnt.msgs, cnt.bytes, + cnt.msgs_dr_ok, cnt.last_offset, cnt.msgs_dr_err, + t_total / 1000, + ((cnt.msgs_dr_ok * 1000000) / t_total), + (float)((cnt.bytes_dr_ok) / (float)t_total), + cnt.tx_err, + rk ? rd_kafka_outq_len(rk) : 0, + compression); + } + + } else { + + if (otype & _OTYPE_TAB) { + if (print_header) { + /* First time, print header */ + ROW_START(); + COL_HDR("elapsed"); + COL_HDR("msgs"); + COL_HDR("bytes"); + COL_HDR("rtt"); + COL_HDR("m/s"); + COL_HDR("MB/s"); + COL_HDR("rx_err"); + COL_HDR("offset"); + if (latency_mode) { + COL_HDR("lat_curr"); + COL_HDR("lat_avg"); + COL_HDR("lat_lo"); + COL_HDR("lat_hi"); + } + ROW_END(); + } + + ROW_START(); + COL_PR64("elapsed", t_total / 1000); + COL_PR64("msgs", cnt.msgs); + COL_PR64("bytes", cnt.bytes); + COL_PR64("rtt", cnt.avg_rtt / 1000); + COL_PR64("m/s", + ((cnt.msgs * 1000000) / t_total)); + COL_PRF("MB/s", + (float)((cnt.bytes) / (float)t_total)); + COL_PR64("rx_err", cnt.msgs_dr_err); + COL_PR64("offset", cnt.offset); + if (latency_mode) { + COL_PRF("lat_curr", cnt.latency_last / 1000.0f); + COL_PRF("lat_avg", latency_avg / 1000.0f); + COL_PRF("lat_lo", cnt.latency_lo / 1000.0f); + COL_PRF("lat_hi", cnt.latency_hi / 1000.0f); + } + ROW_END(); + + } + + if (otype & _OTYPE_SUMMARY) { + if (latency_avg >= 1.0f) + extra_of += rd_snprintf(extra+extra_of, + sizeof(extra)-extra_of, + ", latency " + "curr/avg/lo/hi " + "%.2f/%.2f/%.2f/%.2fms", + cnt.latency_last / 1000.0f, + latency_avg / 1000.0f, + cnt.latency_lo / 1000.0f, + cnt.latency_hi / 1000.0f) +; + printf("%% %"PRIu64" messages (%"PRIu64" bytes) " + "consumed in %"PRIu64"ms: %"PRIu64" msgs/s " + "(%.02f MB/s)" + "%s\n", + cnt.msgs, cnt.bytes, + t_total / 1000, + ((cnt.msgs * 1000000) / t_total), + (float)((cnt.bytes) / (float)t_total), + extra); + } + + if (incremental_mode && now > cnt.t_last) { + uint64_t i_msgs = cnt.msgs - cnt.msgs_last; + uint64_t i_bytes = cnt.bytes - cnt.bytes_last; + uint64_t i_time = cnt.t_last ? now - cnt.t_last : 0; + + printf("%% INTERVAL: %"PRIu64" messages " + "(%"PRIu64" bytes) " + "consumed in %"PRIu64"ms: %"PRIu64" msgs/s " + "(%.02f MB/s)" + "%s\n", + i_msgs, i_bytes, + i_time / 1000, + ((i_msgs * 1000000) / i_time), + (float)((i_bytes) / (float)i_time), + extra); + + } + } + + cnt.t_last = now; + cnt.msgs_last = cnt.msgs; + cnt.bytes_last = cnt.bytes; +} + + +static void sig_usr1 (int sig) { + rd_kafka_dump(stdout, global_rk); +} + + +/** + * @brief Read config from file + * @returns -1 on error, else 0. + */ +static int read_conf_file (rd_kafka_conf_t *conf, + rd_kafka_topic_conf_t *tconf, const char *path) { + FILE *fp; + char buf[512]; + int line = 0; + char errstr[512]; + + if (!(fp = fopen(path, "r"))) { + fprintf(stderr, "%% Failed to open %s: %s\n", + path, strerror(errno)); + return -1; + } + + while (fgets(buf, sizeof(buf), fp)) { + char *s = buf; + char *t; + rd_kafka_conf_res_t r = RD_KAFKA_CONF_UNKNOWN; + + line++; + + while (isspace((int)*s)) + s++; + + if (!*s || *s == '#') + continue; + + if ((t = strchr(buf, '\n'))) + *t = '\0'; + + t = strchr(buf, '='); + if (!t || t == s || !*(t+1)) { + fprintf(stderr, "%% %s:%d: expected key=value\n", + path, line); + fclose(fp); + return -1; + } + + *(t++) = '\0'; + + /* Try property on topic config first */ + if (tconf) + r = rd_kafka_topic_conf_set(tconf, s, t, + errstr, sizeof(errstr)); + + /* Try global config */ + if (r == RD_KAFKA_CONF_UNKNOWN) + r = rd_kafka_conf_set(conf, s, t, + errstr, sizeof(errstr)); + + if (r == RD_KAFKA_CONF_OK) + continue; + + fprintf(stderr, "%% %s:%d: %s=%s: %s\n", + path, line, s, t, errstr); + fclose(fp); + return -1; + } + + fclose(fp); + + return 0; +} + + +static rd_kafka_resp_err_t do_produce (rd_kafka_t *rk, + rd_kafka_topic_t *rkt, int32_t partition, + int msgflags, + void *payload, size_t size, + const void *key, size_t key_size, + const rd_kafka_headers_t *hdrs) { + + /* Send/Produce message. */ + if (hdrs) { + rd_kafka_headers_t *hdrs_copy; + rd_kafka_resp_err_t err; + + hdrs_copy = rd_kafka_headers_copy(hdrs); + + err = rd_kafka_producev( + rk, + RD_KAFKA_V_RKT(rkt), + RD_KAFKA_V_PARTITION(partition), + RD_KAFKA_V_MSGFLAGS(msgflags), + RD_KAFKA_V_VALUE(payload, size), + RD_KAFKA_V_KEY(key, key_size), + RD_KAFKA_V_HEADERS(hdrs_copy), + RD_KAFKA_V_END); + + if (err) + rd_kafka_headers_destroy(hdrs_copy); + + return err; + + } else { + if (rd_kafka_produce(rkt, partition, msgflags, payload, size, + key, key_size, NULL) == -1) + return rd_kafka_last_error(); + } + + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + + +int main (int argc, char **argv) { + char *brokers = NULL; + char mode = 'C'; + char *topic = NULL; + const char *key = NULL; + int *partitions = NULL; + int opt; + int sendflags = 0; + char *msgpattern = "librdkafka_performance testing!"; + int msgsize = (int)strlen(msgpattern); + const char *debug = NULL; + rd_ts_t now; + char errstr[512]; + uint64_t seq = 0; + int seed = (int)time(NULL); + rd_kafka_t *rk; + rd_kafka_topic_t *rkt; + rd_kafka_conf_t *conf; + rd_kafka_topic_conf_t *topic_conf; + rd_kafka_queue_t *rkqu = NULL; + const char *compression = "no"; + int64_t start_offset = 0; + int batch_size = 0; + int idle = 0; + const char *stats_cmd = NULL; + char *stats_intvlstr = NULL; + char tmp[128]; + char *tmp2; + int otype = _OTYPE_SUMMARY; + double dtmp; + int rate_sleep = 0; + rd_kafka_topic_partition_list_t *topics; + int exitcode = 0; + rd_kafka_headers_t *hdrs = NULL; + rd_kafka_resp_err_t err; + + /* Kafka configuration */ + conf = rd_kafka_conf_new(); + rd_kafka_conf_set_error_cb(conf, err_cb); + rd_kafka_conf_set_throttle_cb(conf, throttle_cb); + rd_kafka_conf_set_offset_commit_cb(conf, offset_commit_cb); + +#ifdef SIGIO + /* Quick termination */ + rd_snprintf(tmp, sizeof(tmp), "%i", SIGIO); + rd_kafka_conf_set(conf, "internal.termination.signal", tmp, NULL, 0); +#endif + + /* Producer config */ + rd_kafka_conf_set(conf, "queue.buffering.max.messages", "500000", + NULL, 0); + rd_kafka_conf_set(conf, "message.send.max.retries", "3", NULL, 0); + rd_kafka_conf_set(conf, "retry.backoff.ms", "500", NULL, 0); + + /* Consumer config */ + /* Tell rdkafka to (try to) maintain 1M messages + * in its internal receive buffers. This is to avoid + * application -> rdkafka -> broker per-message ping-pong + * latency. + * The larger the local queue, the higher the performance. + * Try other values with: ... -X queued.min.messages=1000 + */ + rd_kafka_conf_set(conf, "queued.min.messages", "1000000", NULL, 0); + rd_kafka_conf_set(conf, "session.timeout.ms", "6000", NULL, 0); + + /* Kafka topic configuration */ + topic_conf = rd_kafka_topic_conf_new(); + rd_kafka_topic_conf_set(topic_conf, "auto.offset.reset", "earliest", + NULL, 0); + + topics = rd_kafka_topic_partition_list_new(1); + + while ((opt = + getopt(argc, argv, + "PCG:t:p:b:s:k:c:fi:MDd:m:S:x:" + "R:a:z:o:X:B:eT:Y:qvIur:lA:OwNHH:")) != -1) { + switch (opt) { + case 'G': + if (rd_kafka_conf_set(conf, "group.id", optarg, + errstr, sizeof(errstr)) != + RD_KAFKA_CONF_OK) { + fprintf(stderr, "%% %s\n", errstr); + exit(1); + } + /* FALLTHRU */ + case 'P': + case 'C': + mode = opt; + break; + case 't': + rd_kafka_topic_partition_list_add(topics, optarg, + RD_KAFKA_PARTITION_UA); + break; + case 'p': + partition_cnt++; + partitions = realloc(partitions, sizeof(*partitions) * partition_cnt); + partitions[partition_cnt-1] = atoi(optarg); + break; + + case 'b': + brokers = optarg; + break; + case 's': + msgsize = atoi(optarg); + break; + case 'k': + key = optarg; + break; + case 'c': + msgcnt = atoi(optarg); + break; + case 'D': + sendflags |= RD_KAFKA_MSG_F_FREE; + break; + case 'i': + dispintvl = atoi(optarg); + break; + case 'm': + msgpattern = optarg; + break; + case 'S': + seq = strtoull(optarg, NULL, 10); + do_seq = 1; + break; + case 'x': + exit_after = atoi(optarg); + break; + case 'R': + seed = atoi(optarg); + break; + case 'a': + if (rd_kafka_topic_conf_set(topic_conf, + "request.required.acks", + optarg, + errstr, sizeof(errstr)) != + RD_KAFKA_CONF_OK) { + fprintf(stderr, "%% %s\n", errstr); + exit(1); + } + break; + case 'B': + batch_size = atoi(optarg); + break; + case 'z': + if (rd_kafka_conf_set(conf, "compression.codec", + optarg, + errstr, sizeof(errstr)) != + RD_KAFKA_CONF_OK) { + fprintf(stderr, "%% %s\n", errstr); + exit(1); + } + compression = optarg; + break; + case 'o': + if (!strcmp(optarg, "end")) + start_offset = RD_KAFKA_OFFSET_END; + else if (!strcmp(optarg, "beginning")) + start_offset = RD_KAFKA_OFFSET_BEGINNING; + else if (!strcmp(optarg, "stored")) + start_offset = RD_KAFKA_OFFSET_STORED; + else { + start_offset = strtoll(optarg, NULL, 10); + + if (start_offset < 0) + start_offset = RD_KAFKA_OFFSET_TAIL(-start_offset); + } + + break; + case 'e': + exit_eof = 1; + break; + case 'd': + debug = optarg; + break; + case 'H': + { + char *name, *val; + size_t name_sz = -1; + + if (!optarg) { + read_hdrs = 1; + break; + } + + name = optarg; + val = strchr(name, '='); + if (val) { + name_sz = (size_t)(val-name); + val++; /* past the '=' */ + } + + if (!hdrs) + hdrs = rd_kafka_headers_new(8); + + err = rd_kafka_header_add(hdrs, name, name_sz, val, -1); + if (err) { + fprintf(stderr, + "%% Failed to add header %s: %s\n", + name, rd_kafka_err2str(err)); + exit(1); + } + + read_hdrs = 1; + } + break; + case 'X': + { + char *name, *val; + rd_kafka_conf_res_t res; + + if (!strcmp(optarg, "list") || + !strcmp(optarg, "help")) { + rd_kafka_conf_properties_show(stdout); + exit(0); + } + + name = optarg; + if (!(val = strchr(name, '='))) { + fprintf(stderr, "%% Expected " + "-X property=value, not %s\n", name); + exit(1); + } + + *val = '\0'; + val++; + + if (!strcmp(name, "file")) { + if (read_conf_file(conf, topic_conf, val) == -1) + exit(1); + break; + } + + res = RD_KAFKA_CONF_UNKNOWN; + /* Try "topic." prefixed properties on topic + * conf first, and then fall through to global if + * it didnt match a topic configuration property. */ + if (!strncmp(name, "topic.", strlen("topic."))) + res = rd_kafka_topic_conf_set(topic_conf, + name+ + strlen("topic."), + val, + errstr, + sizeof(errstr)); + + if (res == RD_KAFKA_CONF_UNKNOWN) + res = rd_kafka_conf_set(conf, name, val, + errstr, sizeof(errstr)); + + if (res != RD_KAFKA_CONF_OK) { + fprintf(stderr, "%% %s\n", errstr); + exit(1); + } + } + break; + + case 'T': + stats_intvlstr = optarg; + break; + case 'Y': + stats_cmd = optarg; + break; + + case 'q': + verbosity--; + break; + + case 'v': + verbosity++; + break; + + case 'I': + idle = 1; + break; + + case 'u': + otype = _OTYPE_TAB; + verbosity--; /* remove some fluff */ + break; + + case 'r': + dtmp = strtod(optarg, &tmp2); + if (tmp2 == optarg || + (dtmp >= -0.001 && dtmp <= 0.001)) { + fprintf(stderr, "%% Invalid rate: %s\n", + optarg); + exit(1); + } + + rate_sleep = (int)(1000000.0 / dtmp); + break; + + case 'l': + latency_mode = 1; + break; + + case 'A': + if (!(latency_fp = fopen(optarg, "w"))) { + fprintf(stderr, + "%% Cant open %s: %s\n", + optarg, strerror(errno)); + exit(1); + } + break; + + case 'O': + if (rd_kafka_topic_conf_set(topic_conf, + "produce.offset.report", + "true", + errstr, sizeof(errstr)) != + RD_KAFKA_CONF_OK) { + fprintf(stderr, "%% %s\n", errstr); + exit(1); + } + report_offset = 1; + break; + + case 'M': + incremental_mode = 1; + break; + + case 'N': + with_dr = 0; + break; + + default: + fprintf(stderr, "Unknown option: %c\n", opt); + goto usage; + } + } + + if (topics->cnt == 0 || optind != argc) { + if (optind < argc) + fprintf(stderr, "Unknown argument: %s\n", argv[optind]); + usage: + fprintf(stderr, + "Usage: %s [-C|-P] -t <topic> " + "[-p <partition>] [-b <broker,broker..>] [options..]\n" + "\n" + "librdkafka version %s (0x%08x)\n" + "\n" + " Options:\n" + " -C | -P | Consumer or Producer mode\n" + " -G <groupid> High-level Kafka Consumer mode\n" + " -t <topic> Topic to consume / produce\n" + " -p <num> Partition (defaults to random). " + "Multiple partitions are allowed in -C consumer mode.\n" + " -M Print consumer interval stats\n" + " -b <brokers> Broker address list (host[:port],..)\n" + " -s <size> Message size (producer)\n" + " -k <key> Message key (producer)\n" + " -H <name[=value]> Add header to message (producer)\n" + " -H Read message headers (consumer)\n" + " -c <cnt> Messages to transmit/receive\n" + " -x <cnt> Hard exit after transmitting <cnt> messages (producer)\n" + " -D Copy/Duplicate data buffer (producer)\n" + " -i <ms> Display interval\n" + " -m <msg> Message payload pattern\n" + " -S <start> Send a sequence number starting at " + "<start> as payload\n" + " -R <seed> Random seed value (defaults to time)\n" + " -a <acks> Required acks (producer): " + "-1, 0, 1, >1\n" + " -B <size> Consume batch size (# of msgs)\n" + " -z <codec> Enable compression:\n" + " none|gzip|snappy\n" + " -o <offset> Start offset (consumer)\n" + " beginning, end, NNNNN or -NNNNN\n" + " -d [facs..] Enable debugging contexts:\n" + " %s\n" + " -X <prop=name> Set arbitrary librdkafka " + "configuration property\n" + " Properties prefixed with \"topic.\" " + "will be set on topic object.\n" + " Use '-X list' to see the full list\n" + " of supported properties.\n" + " -X file=<path> Read config from file.\n" + " -T <intvl> Enable statistics from librdkafka at " + "specified interval (ms)\n" + " -Y <command> Pipe statistics to <command>\n" + " -I Idle: dont produce any messages\n" + " -q Decrease verbosity\n" + " -v Increase verbosity (default 1)\n" + " -u Output stats in table format\n" + " -r <rate> Producer msg/s limit\n" + " -l Latency measurement.\n" + " Needs two matching instances, one\n" + " consumer and one producer, both\n" + " running with the -l switch.\n" + " -l Producer: per-message latency stats\n" + " -A <file> Write per-message latency stats to " + "<file>. Requires -l\n" + " -O Report produced offset (producer)\n" + " -N No delivery reports (producer)\n" + "\n" + " In Consumer mode:\n" + " consumes messages and prints thruput\n" + " If -B <..> is supplied the batch consumer\n" + " mode is used, else the callback mode is used.\n" + "\n" + " In Producer mode:\n" + " writes messages of size -s <..> and prints thruput\n" + "\n", + argv[0], + rd_kafka_version_str(), rd_kafka_version(), + RD_KAFKA_DEBUG_CONTEXTS); + exit(1); + } + + + dispintvl *= 1000; /* us */ + + if (verbosity > 1) + printf("%% Using random seed %i, verbosity level %i\n", + seed, verbosity); + srand(seed); + signal(SIGINT, stop); +#ifdef SIGUSR1 + signal(SIGUSR1, sig_usr1); +#endif + + + if (debug && + rd_kafka_conf_set(conf, "debug", debug, errstr, sizeof(errstr)) != + RD_KAFKA_CONF_OK) { + printf("%% Debug configuration failed: %s: %s\n", + errstr, debug); + exit(1); + } + + /* Always enable stats (for RTT extraction), and if user supplied + * the -T <intvl> option we let her take part of the stats aswell. */ + rd_kafka_conf_set_stats_cb(conf, stats_cb); + + if (!stats_intvlstr) { + /* if no user-desired stats, adjust stats interval + * to the display interval. */ + rd_snprintf(tmp, sizeof(tmp), "%"PRId64, dispintvl / 1000); + } + + if (rd_kafka_conf_set(conf, "statistics.interval.ms", + stats_intvlstr ? stats_intvlstr : tmp, + errstr, sizeof(errstr)) != + RD_KAFKA_CONF_OK) { + fprintf(stderr, "%% %s\n", errstr); + exit(1); + } + + if (latency_mode) + do_seq = 0; + + if (stats_intvlstr) { + /* User enabled stats (-T) */ + +#ifndef _MSC_VER + if (stats_cmd) { + if (!(stats_fp = popen(stats_cmd, "we"))) { + fprintf(stderr, + "%% Failed to start stats command: " + "%s: %s", stats_cmd, strerror(errno)); + exit(1); + } + } else +#endif + stats_fp = stdout; + } + + if (msgcnt != -1) + forever = 0; + + topic = topics->elems[0].topic; + + if (mode == 'P') { + /* + * Producer + */ + char *sbuf; + char *pbuf; + int outq; + int keylen = key ? (int)strlen(key) : 0; + off_t rof = 0; + size_t plen = strlen(msgpattern); + int partition = partitions ? partitions[0] : + RD_KAFKA_PARTITION_UA; + + if (latency_mode) { + int minlen = (int)(strlen("LATENCY:") + + strlen("18446744073709551615 ")+1); + msgsize = RD_MAX(minlen, msgsize); + sendflags |= RD_KAFKA_MSG_F_COPY; + } else if (do_seq) { + int minlen = (int)strlen("18446744073709551615 ")+1; + if (msgsize < minlen) + msgsize = minlen; + + /* Force duplication of payload */ + sendflags |= RD_KAFKA_MSG_F_FREE; + } + + sbuf = malloc(msgsize); + + /* Copy payload content to new buffer */ + while (rof < msgsize) { + size_t xlen = RD_MIN((size_t)msgsize-rof, plen); + memcpy(sbuf+rof, msgpattern, xlen); + rof += (off_t)xlen; + } + + if (msgcnt == -1) + printf("%% Sending messages of size %i bytes\n", + msgsize); + else + printf("%% Sending %i messages of size %i bytes\n", + msgcnt, msgsize); + + if (with_dr) + rd_kafka_conf_set_dr_msg_cb(conf, msg_delivered); + + /* Create Kafka handle */ + if (!(rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, + errstr, sizeof(errstr)))) { + fprintf(stderr, + "%% Failed to create Kafka producer: %s\n", + errstr); + exit(1); + } + + global_rk = rk; + + /* Add broker(s) */ + if (brokers && rd_kafka_brokers_add(rk, brokers) < 1) { + fprintf(stderr, "%% No valid brokers specified\n"); + exit(1); + } + + /* Explicitly create topic to avoid per-msg lookups. */ + rkt = rd_kafka_topic_new(rk, topic, topic_conf); + + + if (rate_sleep && verbosity >= 2) + fprintf(stderr, + "%% Inter message rate limiter sleep %ius\n", + rate_sleep); + + dr_disp_div = msgcnt / 50; + if (dr_disp_div == 0) + dr_disp_div = 10; + + cnt.t_start = cnt.t_last = rd_clock(); + + msgs_wait_produce_cnt = msgcnt; + + while (run && (msgcnt == -1 || (int)cnt.msgs < msgcnt)) { + /* Send/Produce message. */ + + if (idle) { + rd_kafka_poll(rk, 1000); + continue; + } + + if (latency_mode) { + rd_snprintf(sbuf, msgsize-1, + "LATENCY:%"PRIu64, wall_clock()); + } else if (do_seq) { + rd_snprintf(sbuf, + msgsize-1, "%"PRIu64": ", seq); + seq++; + } + + if (sendflags & RD_KAFKA_MSG_F_FREE) { + /* Duplicate memory */ + pbuf = malloc(msgsize); + memcpy(pbuf, sbuf, msgsize); + } else + pbuf = sbuf; + + if (msgsize == 0) + pbuf = NULL; + + cnt.tx++; + while (run && + (err = do_produce(rk, rkt, partition, sendflags, + pbuf, msgsize, + key, keylen, hdrs))) { + if (err == RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION) + printf("%% No such partition: " + "%"PRId32"\n", partition); + else if (verbosity >= 3 || + (err != RD_KAFKA_RESP_ERR__QUEUE_FULL && verbosity >= 1)) + printf("%% produce error: %s%s\n", + rd_kafka_err2str(err), + err == RD_KAFKA_RESP_ERR__QUEUE_FULL ? + " (backpressure)" : ""); + + cnt.tx_err++; + if (err != RD_KAFKA_RESP_ERR__QUEUE_FULL) { + run = 0; + break; + } + now = rd_clock(); + if (verbosity >= 2 && + cnt.t_enobufs_last + dispintvl <= now) { + printf("%% Backpressure %i " + "(tx %"PRIu64", " + "txerr %"PRIu64")\n", + rd_kafka_outq_len(rk), + cnt.tx, cnt.tx_err); + cnt.t_enobufs_last = now; + } + + /* Poll to handle delivery reports */ + rd_kafka_poll(rk, 10); + + print_stats(rk, mode, otype, compression); + } + + msgs_wait_cnt++; + if (msgs_wait_produce_cnt != -1) + msgs_wait_produce_cnt--; + cnt.msgs++; + cnt.bytes += msgsize; + + if (rate_sleep) { + if (rate_sleep > 100) { +#ifdef _MSC_VER + Sleep(rate_sleep / 1000); +#else + usleep(rate_sleep); +#endif + } else { + rd_ts_t next = rd_clock() + rate_sleep; + while (next > rd_clock()) + ; + } + } + + /* Must poll to handle delivery reports */ + rd_kafka_poll(rk, 0); + + print_stats(rk, mode, otype, compression); + } + + forever = 0; + if (verbosity >= 2) + printf("%% All messages produced, " + "now waiting for %li deliveries\n", + msgs_wait_cnt); + + /* Wait for messages to be delivered */ + while (run && rd_kafka_poll(rk, 1000) != -1) + print_stats(rk, mode, otype, compression); + + + outq = rd_kafka_outq_len(rk); + if (verbosity >= 2) + printf("%% %i messages in outq\n", outq); + cnt.msgs -= outq; + cnt.bytes -= msgsize * outq; + + cnt.t_end = t_end; + + if (cnt.tx_err > 0) + printf("%% %"PRIu64" backpressures for %"PRIu64 + " produce calls: %.3f%% backpressure rate\n", + cnt.tx_err, cnt.tx, + ((double)cnt.tx_err / (double)cnt.tx) * 100.0); + + /* Destroy topic */ + rd_kafka_topic_destroy(rkt); + + /* Destroy the handle */ + rd_kafka_destroy(rk); + global_rk = rk = NULL; + + free(sbuf); + + exitcode = cnt.msgs == cnt.msgs_dr_ok ? 0 : 1; + + } else if (mode == 'C') { + /* + * Consumer + */ + + rd_kafka_message_t **rkmessages = NULL; + size_t i = 0; + + /* Create Kafka handle */ + if (!(rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, + errstr, sizeof(errstr)))) { + fprintf(stderr, + "%% Failed to create Kafka consumer: %s\n", + errstr); + exit(1); + } + + global_rk = rk; + + /* Add broker(s) */ + if (brokers && rd_kafka_brokers_add(rk, brokers) < 1) { + fprintf(stderr, "%% No valid brokers specified\n"); + exit(1); + } + + /* Create topic to consume from */ + rkt = rd_kafka_topic_new(rk, topic, topic_conf); + + /* Batch consumer */ + if (batch_size) + rkmessages = malloc(sizeof(*rkmessages) * batch_size); + + /* Start consuming */ + rkqu = rd_kafka_queue_new(rk); + for (i=0 ; i<(size_t)partition_cnt ; ++i) { + const int r = rd_kafka_consume_start_queue(rkt, + partitions[i], start_offset, rkqu); + + if (r == -1) { + fprintf(stderr, "%% Error creating queue: %s\n", + rd_kafka_err2str(rd_kafka_last_error())); + exit(1); + } + } + + while (run && (msgcnt == -1 || msgcnt > (int)cnt.msgs)) { + /* Consume messages. + * A message may either be a real message, or + * an error signaling (if rkmessage->err is set). + */ + uint64_t fetch_latency; + ssize_t r; + + fetch_latency = rd_clock(); + + if (batch_size) { + int i; + int partition = partitions ? partitions[0] : + RD_KAFKA_PARTITION_UA; + + /* Batch fetch mode */ + r = rd_kafka_consume_batch(rkt, partition, + 1000, + rkmessages, + batch_size); + if (r != -1) { + for (i = 0 ; i < r ; i++) { + msg_consume(rkmessages[i], + NULL); + rd_kafka_message_destroy( + rkmessages[i]); + } + } + } else { + /* Queue mode */ + r = rd_kafka_consume_callback_queue(rkqu, 1000, + msg_consume, + NULL); + } + + cnt.t_fetch_latency += rd_clock() - fetch_latency; + if (r == -1) + fprintf(stderr, "%% Error: %s\n", + rd_kafka_err2str(rd_kafka_last_error())); + + print_stats(rk, mode, otype, compression); + + /* Poll to handle stats callbacks */ + rd_kafka_poll(rk, 0); + } + cnt.t_end = rd_clock(); + + /* Stop consuming */ + for (i=0 ; i<(size_t)partition_cnt ; ++i) { + int r = rd_kafka_consume_stop(rkt, (int32_t)i); + if (r == -1) { + fprintf(stderr, + "%% Error in consume_stop: %s\n", + rd_kafka_err2str(rd_kafka_last_error())); + } + } + rd_kafka_queue_destroy(rkqu); + + /* Destroy topic */ + rd_kafka_topic_destroy(rkt); + + if (batch_size) + free(rkmessages); + + /* Destroy the handle */ + rd_kafka_destroy(rk); + + global_rk = rk = NULL; + + } else if (mode == 'G') { + /* + * High-level balanced Consumer + */ + rd_kafka_resp_err_t err; + + rd_kafka_conf_set_rebalance_cb(conf, rebalance_cb); + rd_kafka_conf_set_default_topic_conf(conf, topic_conf); + + /* Create Kafka handle */ + if (!(rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, + errstr, sizeof(errstr)))) { + fprintf(stderr, + "%% Failed to create Kafka consumer: %s\n", + errstr); + exit(1); + } + + /* Forward all events to consumer queue */ + rd_kafka_poll_set_consumer(rk); + + global_rk = rk; + + /* Add broker(s) */ + if (brokers && rd_kafka_brokers_add(rk, brokers) < 1) { + fprintf(stderr, "%% No valid brokers specified\n"); + exit(1); + } + + err = rd_kafka_subscribe(rk, topics); + if (err) { + fprintf(stderr, "%% Subscribe failed: %s\n", + rd_kafka_err2str(err)); + exit(1); + } + fprintf(stderr, "%% Waiting for group rebalance..\n"); + + while (run && (msgcnt == -1 || msgcnt > (int)cnt.msgs)) { + /* Consume messages. + * A message may either be a real message, or + * an event (if rkmessage->err is set). + */ + rd_kafka_message_t *rkmessage; + uint64_t fetch_latency; + + fetch_latency = rd_clock(); + + rkmessage = rd_kafka_consumer_poll(rk, 1000); + if (rkmessage) { + msg_consume(rkmessage, NULL); + rd_kafka_message_destroy(rkmessage); + } + + cnt.t_fetch_latency += rd_clock() - fetch_latency; + + print_stats(rk, mode, otype, compression); + } + cnt.t_end = rd_clock(); + + err = rd_kafka_consumer_close(rk); + if (err) + fprintf(stderr, "%% Failed to close consumer: %s\n", + rd_kafka_err2str(err)); + + rd_kafka_destroy(rk); + } + + if (hdrs) + rd_kafka_headers_destroy(hdrs); + + print_stats(NULL, mode, otype|_OTYPE_FORCE, compression); + + if (cnt.t_fetch_latency && cnt.msgs) + printf("%% Average application fetch latency: %"PRIu64"us\n", + cnt.t_fetch_latency / cnt.msgs); + + if (latency_fp) + fclose(latency_fp); + + if (stats_fp) { +#ifndef _MSC_VER + pclose(stats_fp); +#endif + stats_fp = NULL; + } + + if (partitions) + free(partitions); + + rd_kafka_topic_partition_list_destroy(topics); + + /* Let background threads clean up and terminate cleanly. */ + rd_kafka_wait_destroyed(2000); + + return exitcode; +}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/rdkafka_simple_producer.c ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.4/examples/rdkafka_simple_producer.c b/thirdparty/librdkafka-0.11.4/examples/rdkafka_simple_producer.c new file mode 100644 index 0000000..a353d01 --- /dev/null +++ b/thirdparty/librdkafka-0.11.4/examples/rdkafka_simple_producer.c @@ -0,0 +1,260 @@ +/* + * librdkafka - Apache Kafka C library + * + * Copyright (c) 2017, 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. + */ + +/** + * Simple Apache Kafka producer + * using the Kafka driver from librdkafka + * (https://github.com/edenhill/librdkafka) + */ + +#include <stdio.h> +#include <signal.h> +#include <string.h> + + +/* Typical include path would be <librdkafka/rdkafka.h>, but this program + * is builtin from within the librdkafka source tree and thus differs. */ +#include "rdkafka.h" + + +static int run = 1; + +/** + * @brief Signal termination of program + */ +static void stop (int sig) { + run = 0; + fclose(stdin); /* abort fgets() */ +} + + +/** + * @brief Message delivery report callback. + * + * This callback is called exactly once per message, indicating if + * the message was succesfully delivered + * (rkmessage->err == RD_KAFKA_RESP_ERR_NO_ERROR) or permanently + * failed delivery (rkmessage->err != RD_KAFKA_RESP_ERR_NO_ERROR). + * + * The callback is triggered from rd_kafka_poll() and executes on + * the application's thread. + */ +static void dr_msg_cb (rd_kafka_t *rk, + const rd_kafka_message_t *rkmessage, void *opaque) { + if (rkmessage->err) + fprintf(stderr, "%% Message delivery failed: %s\n", + rd_kafka_err2str(rkmessage->err)); + else + fprintf(stderr, + "%% Message delivered (%zd bytes, " + "partition %"PRId32")\n", + rkmessage->len, rkmessage->partition); + + /* The rkmessage is destroyed automatically by librdkafka */ +} + + + +int main (int argc, char **argv) { + rd_kafka_t *rk; /* Producer instance handle */ + rd_kafka_topic_t *rkt; /* Topic object */ + rd_kafka_conf_t *conf; /* Temporary configuration object */ + char errstr[512]; /* librdkafka API error reporting buffer */ + char buf[512]; /* Message value temporary buffer */ + const char *brokers; /* Argument: broker list */ + const char *topic; /* Argument: topic to produce to */ + + /* + * Argument validation + */ + if (argc != 3) { + fprintf(stderr, "%% Usage: %s <broker> <topic>\n", argv[0]); + return 1; + } + + brokers = argv[1]; + topic = argv[2]; + + + /* + * Create Kafka client configuration place-holder + */ + conf = rd_kafka_conf_new(); + + /* Set bootstrap broker(s) as a comma-separated list of + * host or host:port (default port 9092). + * librdkafka will use the bootstrap brokers to acquire the full + * set of brokers from the cluster. */ + if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers, + errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) { + fprintf(stderr, "%s\n", errstr); + return 1; + } + + /* Set the delivery report callback. + * This callback will be called once per message to inform + * the application if delivery succeeded or failed. + * See dr_msg_cb() above. */ + rd_kafka_conf_set_dr_msg_cb(conf, dr_msg_cb); + + + /* + * Create producer instance. + * + * NOTE: rd_kafka_new() takes ownership of the conf object + * and the application must not reference it again after + * this call. + */ + rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr)); + if (!rk) { + fprintf(stderr, + "%% Failed to create new producer: %s\n", errstr); + return 1; + } + + + /* Create topic object that will be reused for each message + * produced. + * + * Both the producer instance (rd_kafka_t) and topic objects (topic_t) + * are long-lived objects that should be reused as much as possible. + */ + rkt = rd_kafka_topic_new(rk, topic, NULL); + if (!rkt) { + fprintf(stderr, "%% Failed to create topic object: %s\n", + rd_kafka_err2str(rd_kafka_last_error())); + rd_kafka_destroy(rk); + return 1; + } + + /* Signal handler for clean shutdown */ + signal(SIGINT, stop); + + fprintf(stderr, + "%% Type some text and hit enter to produce message\n" + "%% Or just hit enter to only serve delivery reports\n" + "%% Press Ctrl-C or Ctrl-D to exit\n"); + + while (run && fgets(buf, sizeof(buf), stdin)) { + size_t len = strlen(buf); + + if (buf[len-1] == '\n') /* Remove newline */ + buf[--len] = '\0'; + + if (len == 0) { + /* Empty line: only serve delivery reports */ + rd_kafka_poll(rk, 0/*non-blocking */); + continue; + } + + /* + * Send/Produce message. + * This is an asynchronous call, on success it will only + * enqueue the message on the internal producer queue. + * The actual delivery attempts to the broker are handled + * by background threads. + * The previously registered delivery report callback + * (dr_msg_cb) is used to signal back to the application + * when the message has been delivered (or failed). + */ + retry: + if (rd_kafka_produce( + /* Topic object */ + rkt, + /* Use builtin partitioner to select partition*/ + RD_KAFKA_PARTITION_UA, + /* Make a copy of the payload. */ + RD_KAFKA_MSG_F_COPY, + /* Message payload (value) and length */ + buf, len, + /* Optional key and its length */ + NULL, 0, + /* Message opaque, provided in + * delivery report callback as + * msg_opaque. */ + NULL) == -1) { + /** + * Failed to *enqueue* message for producing. + */ + fprintf(stderr, + "%% Failed to produce to topic %s: %s\n", + rd_kafka_topic_name(rkt), + rd_kafka_err2str(rd_kafka_last_error())); + + /* Poll to handle delivery reports */ + if (rd_kafka_last_error() == + RD_KAFKA_RESP_ERR__QUEUE_FULL) { + /* If the internal queue is full, wait for + * messages to be delivered and then retry. + * The internal queue represents both + * messages to be sent and messages that have + * been sent or failed, awaiting their + * delivery report callback to be called. + * + * The internal queue is limited by the + * configuration property + * queue.buffering.max.messages */ + rd_kafka_poll(rk, 1000/*block for max 1000ms*/); + goto retry; + } + } else { + fprintf(stderr, "%% Enqueued message (%zd bytes) " + "for topic %s\n", + len, rd_kafka_topic_name(rkt)); + } + + + /* A producer application should continually serve + * the delivery report queue by calling rd_kafka_poll() + * at frequent intervals. + * Either put the poll call in your main loop, or in a + * dedicated thread, or call it after every + * rd_kafka_produce() call. + * Just make sure that rd_kafka_poll() is still called + * during periods where you are not producing any messages + * to make sure previously produced messages have their + * delivery report callback served (and any other callbacks + * you register). */ + rd_kafka_poll(rk, 0/*non-blocking*/); + } + + + /* Wait for final messages to be delivered or fail. + * rd_kafka_flush() is an abstraction over rd_kafka_poll() which + * waits for all messages to be delivered. */ + fprintf(stderr, "%% Flushing final messages..\n"); + rd_kafka_flush(rk, 10*1000 /* wait for max 10 seconds */); + + /* Destroy topic object */ + rd_kafka_topic_destroy(rkt); + + /* Destroy the producer instance */ + rd_kafka_destroy(rk); + + return 0; +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/rdkafka_zookeeper_example.c ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.4/examples/rdkafka_zookeeper_example.c b/thirdparty/librdkafka-0.11.4/examples/rdkafka_zookeeper_example.c new file mode 100644 index 0000000..2f9a61e --- /dev/null +++ b/thirdparty/librdkafka-0.11.4/examples/rdkafka_zookeeper_example.c @@ -0,0 +1,728 @@ +/* + * librdkafka - Apache Kafka C library + * + * Copyright (c) 2012, 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. + */ + +/** + * Apache Kafka consumer & producer example programs + * using the Kafka driver from librdkafka + * (https://github.com/edenhill/librdkafka) + */ + +#include <ctype.h> +#include <signal.h> +#include <string.h> +#include <unistd.h> +#include <stdlib.h> +#include <syslog.h> +#include <sys/time.h> +#include <errno.h> + +/* Typical include path would be <librdkafka/rdkafka.h>, but this program + * is builtin from within the librdkafka source tree and thus differs. */ +#include "rdkafka.h" /* for Kafka driver */ + +#include <zookeeper.h> +#include <zookeeper.jute.h> +#include <jansson.h> + +#define BROKER_PATH "/brokers/ids" + +static int run = 1; +static rd_kafka_t *rk; +static int exit_eof = 0; +static int quiet = 0; +static enum { + OUTPUT_HEXDUMP, + OUTPUT_RAW, +} output = OUTPUT_HEXDUMP; + +static void stop (int sig) { + run = 0; + fclose(stdin); /* abort fgets() */ +} + + +static void hexdump (FILE *fp, const char *name, const void *ptr, size_t len) { + const char *p = (const char *)ptr; + int of = 0; + + + if (name) + fprintf(fp, "%s hexdump (%zd bytes):\n", name, len); + + for (of = 0 ; of < len ; of += 16) { + char hexen[16*3+1]; + char charen[16+1]; + int hof = 0; + + int cof = 0; + int i; + + for (i = of ; i < of + 16 && i < len ; i++) { + hof += sprintf(hexen+hof, "%02x ", p[i] & 0xff); + cof += sprintf(charen+cof, "%c", + isprint((int)p[i]) ? p[i] : '.'); + } + fprintf(fp, "%08x: %-48s %-16s\n", + of, hexen, charen); + } +} + +/** + * Kafka logger callback (optional) + */ +static void logger (const rd_kafka_t *rk, int level, + const char *fac, const char *buf) { + struct timeval tv; + gettimeofday(&tv, NULL); + fprintf(stderr, "%u.%03u RDKAFKA-%i-%s: %s: %s\n", + (int)tv.tv_sec, (int)(tv.tv_usec / 1000), + level, fac, rd_kafka_name(rk), buf); +} + +/** + * Message delivery report callback. + * Called once for each message. + * See rdkafka.h for more information. + */ +static void msg_delivered (rd_kafka_t *rk, + void *payload, size_t len, + int error_code, + void *opaque, void *msg_opaque) { + + if (error_code) + fprintf(stderr, "%% Message delivery failed: %s\n", + rd_kafka_err2str(error_code)); + else if (!quiet) + fprintf(stderr, "%% Message delivered (%zd bytes)\n", len); +} + + +static void msg_consume (rd_kafka_message_t *rkmessage, + void *opaque) { + if (rkmessage->err) { + if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { + fprintf(stderr, + "%% Consumer reached end of %s [%"PRId32"] " + "message queue at offset %"PRId64"\n", + rd_kafka_topic_name(rkmessage->rkt), + rkmessage->partition, rkmessage->offset); + + if (exit_eof) + run = 0; + + return; + } + + fprintf(stderr, "%% Consume error for topic \"%s\" [%"PRId32"] " + "offset %"PRId64": %s\n", + rd_kafka_topic_name(rkmessage->rkt), + rkmessage->partition, + rkmessage->offset, + rd_kafka_message_errstr(rkmessage)); + return; + } + + if (!quiet) + fprintf(stdout, "%% Message (offset %"PRId64", %zd bytes):\n", + rkmessage->offset, rkmessage->len); + + if (rkmessage->key_len) { + if (output == OUTPUT_HEXDUMP) + hexdump(stdout, "Message Key", + rkmessage->key, rkmessage->key_len); + else + printf("Key: %.*s\n", + (int)rkmessage->key_len, (char *)rkmessage->key); + } + + if (output == OUTPUT_HEXDUMP) + hexdump(stdout, "Message Payload", + rkmessage->payload, rkmessage->len); + else + printf("%.*s\n", + (int)rkmessage->len, (char *)rkmessage->payload); +} + + +static void metadata_print (const char *topic, + const struct rd_kafka_metadata *metadata) { + int i, j, k; + + printf("Metadata for %s (from broker %"PRId32": %s):\n", + topic ? : "all topics", + metadata->orig_broker_id, + metadata->orig_broker_name); + + + /* Iterate brokers */ + printf(" %i brokers:\n", metadata->broker_cnt); + for (i = 0 ; i < metadata->broker_cnt ; i++) + printf(" broker %"PRId32" at %s:%i\n", + metadata->brokers[i].id, + metadata->brokers[i].host, + metadata->brokers[i].port); + + /* Iterate topics */ + printf(" %i topics:\n", metadata->topic_cnt); + for (i = 0 ; i < metadata->topic_cnt ; i++) { + const struct rd_kafka_metadata_topic *t = &metadata->topics[i]; + printf(" topic \"%s\" with %i partitions:", + t->topic, + t->partition_cnt); + if (t->err) { + printf(" %s", rd_kafka_err2str(t->err)); + if (t->err == RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE) + printf(" (try again)"); + } + printf("\n"); + + /* Iterate topic's partitions */ + for (j = 0 ; j < t->partition_cnt ; j++) { + const struct rd_kafka_metadata_partition *p; + p = &t->partitions[j]; + printf(" partition %"PRId32", " + "leader %"PRId32", replicas: ", + p->id, p->leader); + + /* Iterate partition's replicas */ + for (k = 0 ; k < p->replica_cnt ; k++) + printf("%s%"PRId32, + k > 0 ? ",":"", p->replicas[k]); + + /* Iterate partition's ISRs */ + printf(", isrs: "); + for (k = 0 ; k < p->isr_cnt ; k++) + printf("%s%"PRId32, + k > 0 ? ",":"", p->isrs[k]); + if (p->err) + printf(", %s\n", rd_kafka_err2str(p->err)); + else + printf("\n"); + } + } +} + + +static void set_brokerlist_from_zookeeper(zhandle_t *zzh, char *brokers) +{ + if (zzh) + { + struct String_vector brokerlist; + if (zoo_get_children(zzh, BROKER_PATH, 1, &brokerlist) != ZOK) + { + fprintf(stderr, "No brokers found on path %s\n", BROKER_PATH); + return; + } + + int i; + char *brokerptr = brokers; + for (i = 0; i < brokerlist.count; i++) + { + char path[255], cfg[1024]; + sprintf(path, "/brokers/ids/%s", brokerlist.data[i]); + int len = sizeof(cfg); + zoo_get(zzh, path, 0, cfg, &len, NULL); + + if (len > 0) + { + cfg[len] = '\0'; + json_error_t jerror; + json_t *jobj = json_loads(cfg, 0, &jerror); + if (jobj) + { + json_t *jhost = json_object_get(jobj, "host"); + json_t *jport = json_object_get(jobj, "port"); + + if (jhost && jport) + { + const char *host = json_string_value(jhost); + const int port = json_integer_value(jport); + sprintf(brokerptr, "%s:%d", host, port); + + brokerptr += strlen(brokerptr); + if (i < brokerlist.count - 1) + { + *brokerptr++ = ','; + } + } + json_decref(jobj); + } + } + } + deallocate_String_vector(&brokerlist); + printf("Found brokers %s\n", brokers); + } +} + + +static void watcher(zhandle_t *zh, int type, int state, const char *path, void *watcherCtx) +{ + char brokers[1024]; + if (type == ZOO_CHILD_EVENT && strncmp(path, BROKER_PATH, sizeof(BROKER_PATH) - 1) == 0) + { + brokers[0] = '\0'; + set_brokerlist_from_zookeeper(zh, brokers); + if (brokers[0] != '\0' && rk != NULL) + { + rd_kafka_brokers_add(rk, brokers); + rd_kafka_poll(rk, 10); + } + } +} + + +static zhandle_t* initialize_zookeeper(const char * zookeeper, const int debug) +{ + zhandle_t *zh; + if (debug) + { + zoo_set_debug_level(ZOO_LOG_LEVEL_DEBUG); + } + zh = zookeeper_init(zookeeper, watcher, 10000, 0, 0, 0); + if (zh == NULL) + { + fprintf(stderr, "Zookeeper connection not established."); + exit(1); + } + return zh; +} + + +static void sig_usr1 (int sig) { + rd_kafka_dump(stdout, rk); +} + +int main (int argc, char **argv) { + rd_kafka_topic_t *rkt; + char *zookeeper = "localhost:2181"; + zhandle_t *zh = NULL; + char brokers[1024]; + char mode = 'C'; + char *topic = NULL; + int partition = RD_KAFKA_PARTITION_UA; + int opt; + rd_kafka_conf_t *conf; + rd_kafka_topic_conf_t *topic_conf; + char errstr[512]; + const char *debug = NULL; + int64_t start_offset = 0; + int do_conf_dump = 0; + + memset(brokers, 0, sizeof(brokers)); + quiet = !isatty(STDIN_FILENO); + + /* Kafka configuration */ + conf = rd_kafka_conf_new(); + + /* Topic configuration */ + topic_conf = rd_kafka_topic_conf_new(); + + while ((opt = getopt(argc, argv, "PCLt:p:k:z:qd:o:eX:A")) != -1) { + switch (opt) { + case 'P': + case 'C': + case 'L': + mode = opt; + break; + case 't': + topic = optarg; + break; + case 'p': + partition = atoi(optarg); + break; + case 'k': + zookeeper = optarg; + break; + case 'z': + if (rd_kafka_conf_set(conf, "compression.codec", + optarg, + errstr, sizeof(errstr)) != + RD_KAFKA_CONF_OK) { + fprintf(stderr, "%% %s\n", errstr); + exit(1); + } + break; + case 'o': + if (!strcmp(optarg, "end")) + start_offset = RD_KAFKA_OFFSET_END; + else if (!strcmp(optarg, "beginning")) + start_offset = RD_KAFKA_OFFSET_BEGINNING; + else if (!strcmp(optarg, "stored")) + start_offset = RD_KAFKA_OFFSET_STORED; + else + start_offset = strtoll(optarg, NULL, 10); + break; + case 'e': + exit_eof = 1; + break; + case 'd': + debug = optarg; + break; + case 'q': + quiet = 1; + break; + case 'A': + output = OUTPUT_RAW; + break; + case 'X': + { + char *name, *val; + rd_kafka_conf_res_t res; + + if (!strcmp(optarg, "list") || + !strcmp(optarg, "help")) { + rd_kafka_conf_properties_show(stdout); + exit(0); + } + + if (!strcmp(optarg, "dump")) { + do_conf_dump = 1; + continue; + } + + name = optarg; + if (!(val = strchr(name, '='))) { + fprintf(stderr, "%% Expected " + "-X property=value, not %s\n", name); + exit(1); + } + + *val = '\0'; + val++; + + res = RD_KAFKA_CONF_UNKNOWN; + /* Try "topic." prefixed properties on topic + * conf first, and then fall through to global if + * it didnt match a topic configuration property. */ + if (!strncmp(name, "topic.", strlen("topic."))) + res = rd_kafka_topic_conf_set(topic_conf, + name+ + strlen("topic."), + val, + errstr, + sizeof(errstr)); + + if (res == RD_KAFKA_CONF_UNKNOWN) + res = rd_kafka_conf_set(conf, name, val, + errstr, sizeof(errstr)); + + if (res != RD_KAFKA_CONF_OK) { + fprintf(stderr, "%% %s\n", errstr); + exit(1); + } + } + break; + + default: + goto usage; + } + } + + + if (do_conf_dump) { + const char **arr; + size_t cnt; + int pass; + + for (pass = 0 ; pass < 2 ; pass++) { + int i; + + if (pass == 0) { + arr = rd_kafka_conf_dump(conf, &cnt); + printf("# Global config\n"); + } else { + printf("# Topic config\n"); + arr = rd_kafka_topic_conf_dump(topic_conf, + &cnt); + } + + for (i = 0 ; i < cnt ; i += 2) + printf("%s = %s\n", + arr[i], arr[i+1]); + + printf("\n"); + + rd_kafka_conf_dump_free(arr, cnt); + } + + exit(0); + } + + + if (optind != argc || (mode != 'L' && !topic)) { + usage: + fprintf(stderr, + "Usage: %s -C|-P|-L -t <topic> " + "[-p <partition>] [-b <host1:port1,host2:port2,..>]\n" + "\n" + "librdkafka version %s (0x%08x)\n" + "\n" + " Options:\n" + " -C | -P Consumer or Producer mode\n" + " -L Metadata list mode\n" + " -t <topic> Topic to fetch / produce\n" + " -p <num> Partition (random partitioner)\n" + " -k <zookeepers> Zookeeper address (localhost:2181)\n" + " -z <codec> Enable compression:\n" + " none|gzip|snappy\n" + " -o <offset> Start offset (consumer)\n" + " -e Exit consumer when last message\n" + " in partition has been received.\n" + " -d [facs..] Enable debugging contexts:\n" + " -q Be quiet\n" + " -A Raw payload output (consumer)\n" + " %s\n" + " -X <prop=name> Set arbitrary librdkafka " + "configuration property\n" + " Properties prefixed with \"topic.\" " + "will be set on topic object.\n" + " Use '-X list' to see the full list\n" + " of supported properties.\n" + "\n" + " In Consumer mode:\n" + " writes fetched messages to stdout\n" + " In Producer mode:\n" + " reads messages from stdin and sends to broker\n" + " In List mode:\n" + " queries broker for metadata information, " + "topic is optional.\n" + "\n" + "\n" + "\n", + argv[0], + rd_kafka_version_str(), rd_kafka_version(), + RD_KAFKA_DEBUG_CONTEXTS); + exit(1); + } + + + signal(SIGINT, stop); + signal(SIGUSR1, sig_usr1); + + if (debug && + rd_kafka_conf_set(conf, "debug", debug, errstr, sizeof(errstr)) != + RD_KAFKA_CONF_OK) { + fprintf(stderr, "%% Debug configuration failed: %s: %s\n", + errstr, debug); + exit(1); + } + + /* Set logger */ + rd_kafka_conf_set_log_cb(conf, logger); + + /** Initialize zookeeper */ + zh = initialize_zookeeper(zookeeper, debug != NULL); + + /* Add brokers */ + set_brokerlist_from_zookeeper(zh, brokers); + if (rd_kafka_conf_set(conf, "metadata.broker.list", + brokers, errstr, sizeof(errstr) != + RD_KAFKA_CONF_OK)) { + fprintf(stderr, "%% Failed to set brokers: %s\n", errstr); + exit(1); + } + + if (debug) { + printf("Broker list from zookeeper cluster %s: %s\n", zookeeper, brokers); + } + + if (mode == 'P') { + /* + * Producer + */ + char buf[2048]; + int sendcnt = 0; + + /* Set up a message delivery report callback. + * It will be called once for each message, either on successful + * delivery to broker, or upon failure to deliver to broker. */ + rd_kafka_conf_set_dr_cb(conf, msg_delivered); + + /* Create Kafka handle */ + if (!(rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, + errstr, sizeof(errstr)))) { + fprintf(stderr, + "%% Failed to create new producer: %s\n", + errstr); + exit(1); + } + + /* Create topic */ + rkt = rd_kafka_topic_new(rk, topic, topic_conf); + + if (!quiet) + fprintf(stderr, + "%% Type stuff and hit enter to send\n"); + + while (run && fgets(buf, sizeof(buf), stdin)) { + size_t len = strlen(buf); + if (buf[len-1] == '\n') + buf[--len] = '\0'; + + /* Send/Produce message. */ + if (rd_kafka_produce(rkt, partition, + RD_KAFKA_MSG_F_COPY, + /* Payload and length */ + buf, len, + /* Optional key and its length */ + NULL, 0, + /* Message opaque, provided in + * delivery report callback as + * msg_opaque. */ + NULL) == -1) { + fprintf(stderr, + "%% Failed to produce to topic %s " + "partition %i: %s\n", + rd_kafka_topic_name(rkt), partition, + rd_kafka_err2str( + rd_kafka_errno2err(errno))); + /* Poll to handle delivery reports */ + rd_kafka_poll(rk, 0); + continue; + } + + if (!quiet) + fprintf(stderr, "%% Sent %zd bytes to topic " + "%s partition %i\n", + len, rd_kafka_topic_name(rkt), partition); + sendcnt++; + /* Poll to handle delivery reports */ + rd_kafka_poll(rk, 0); + } + + /* Poll to handle delivery reports */ + rd_kafka_poll(rk, 0); + + /* Wait for messages to be delivered */ + while (run && rd_kafka_outq_len(rk) > 0) + rd_kafka_poll(rk, 100); + + /* Destroy the handle */ + rd_kafka_destroy(rk); + + } else if (mode == 'C') { + /* + * Consumer + */ + + /* Create Kafka handle */ + if (!(rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, + errstr, sizeof(errstr)))) { + fprintf(stderr, + "%% Failed to create new consumer: %s\n", + errstr); + exit(1); + } + + /* Create topic */ + rkt = rd_kafka_topic_new(rk, topic, topic_conf); + + /* Start consuming */ + if (rd_kafka_consume_start(rkt, partition, start_offset) == -1){ + fprintf(stderr, "%% Failed to start consuming: %s\n", + rd_kafka_err2str(rd_kafka_errno2err(errno))); + exit(1); + } + + while (run) { + rd_kafka_message_t *rkmessage; + + /* Consume single message. + * See rdkafka_performance.c for high speed + * consuming of messages. */ + rkmessage = rd_kafka_consume(rkt, partition, 1000); + if (!rkmessage) /* timeout */ + continue; + + msg_consume(rkmessage, NULL); + + /* Return message to rdkafka */ + rd_kafka_message_destroy(rkmessage); + } + + /* Stop consuming */ + rd_kafka_consume_stop(rkt, partition); + + rd_kafka_topic_destroy(rkt); + + rd_kafka_destroy(rk); + + } else if (mode == 'L') { + rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR; + + /* Create Kafka handle */ + if (!(rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, + errstr, sizeof(errstr)))) { + fprintf(stderr, + "%% Failed to create new producer: %s\n", + errstr); + exit(1); + } + + /* Create topic */ + if (topic) + rkt = rd_kafka_topic_new(rk, topic, topic_conf); + else + rkt = NULL; + + while (run) { + const struct rd_kafka_metadata *metadata; + + /* Fetch metadata */ + err = rd_kafka_metadata(rk, rkt ? 0 : 1, rkt, + &metadata, 5000); + if (err != RD_KAFKA_RESP_ERR_NO_ERROR) { + fprintf(stderr, + "%% Failed to acquire metadata: %s\n", + rd_kafka_err2str(err)); + run = 0; + break; + } + + metadata_print(topic, metadata); + + rd_kafka_metadata_destroy(metadata); + run = 0; + } + + /* Destroy the handle */ + rd_kafka_destroy(rk); + + /* Exit right away, dont wait for background cleanup, we haven't + * done anything important anyway. */ + exit(err ? 2 : 0); + } + + /* Let background threads clean up and terminate cleanly. */ + rd_kafka_wait_destroyed(2000); + + /** Free the zookeeper data. */ + zookeeper_close(zh); + + return 0; +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/lds-gen.py ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.4/lds-gen.py b/thirdparty/librdkafka-0.11.4/lds-gen.py new file mode 100755 index 0000000..1136580 --- /dev/null +++ b/thirdparty/librdkafka-0.11.4/lds-gen.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python +# +# +# Generate linker script to only expose symbols of the public API +# + +import sys +import re + + +if __name__ == '__main__': + + funcs = list() + last_line = '' + + for line in sys.stdin: + m = re.match(r'^(\S+.*\s+\**)?(rd_kafka_\S+)\s*\(', line) + if m: + sym = m.group(2) + # Ignore static (unused) functions + m2 = re.match(r'(RD_UNUSED|__attribute__\(\(unused\)\))', last_line) + if not m2: + funcs.append(sym) + last_line = '' + else: + last_line = line + + print('# Automatically generated by lds-gen.py - DO NOT EDIT') + print('{\n global:') + if len(funcs) == 0: + print(' *;') + else: + for f in sorted(funcs): + print(' %s;' % f) + + print('local:\n *;') + + print('};') http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/mainpage.doxy ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.4/mainpage.doxy b/thirdparty/librdkafka-0.11.4/mainpage.doxy new file mode 100644 index 0000000..97f2456 --- /dev/null +++ b/thirdparty/librdkafka-0.11.4/mainpage.doxy @@ -0,0 +1,35 @@ +/** + * @mainpage librdkafka documentation + * + * librdkafka is the Apache Kafka C/C++ client library. + * + * @section intro Introduction + * + * For an introduction and manual to librdkafka see INTRODUCTION.md + * + * @section conf Configuration + * + * librdkafka is highly configurable to meet any deployment demands. + * It is usually safe to leave most configuration properties to their default + * values. + * + * See CONFIGURATION.md for the full list of supported configuration properties. + * + * @remark Application developers are recommended to provide a non-hardcoded + * interface to librdkafka's string based name-value configuration + * property interface, allowing users to configure any librdkafka + * property directly without alterations to the application. + * This allows for seamless upgrades where linking to a new version + * of librdkafka automatically provides new configuration + * based features. + + * + * @section c_api C API + * + * The C API is documented in rdkafka.h + * + * @section cpp_api C++ API + * + * The C++ API is documented in rdkafkacpp.h + */ +
