/*

 * 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 <time.h>

#include <sys/time.h>

#include <getopt.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 */

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;

	size_t 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 < (int)of + 16 && i < (int)len ; i++) {

			hof += sprintf(hexen+hof, "%02x ", p[i] & 0xff);

			cof += sprintf(charen+cof, "%c",

				       isprint((int)p[i]) ? p[i] : '.');

		}

		fprintf(fp, "%08zx: %-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, rk ? rd_kafka_name(rk) : NULL, 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): %.*s\n", len,

			(int)len, (const char *)payload);

}

/**

 * Message delivery report callback using the richer rd_kafka_message_t object.

 */

static void msg_delivered2 (rd_kafka_t *rk,

                            const rd_kafka_message_t *rkmessage, void *opaque) {

	printf("del: %s: offset %"PRId64"\n",

	       rd_kafka_err2str(rkmessage->err), rkmessage->offset);

        if (rkmessage->err)

		fprintf(stderr, "%% Message delivery failed: %s\n",

                        rd_kafka_err2str(rkmessage->err));

	else if (!quiet)

		fprintf(stderr,

                        "%% Message delivered (%zd bytes, offset %"PRId64", "

                        "partition %"PRId32"): %.*s\n",

                        rkmessage->len, rkmessage->offset,

			rkmessage->partition,

			(int)rkmessage->len, (const char *)rkmessage->payload);

}

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));

                if (rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION ||

                    rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)

                        run = 0;

		return;

	}

	if (!quiet) {

		rd_kafka_timestamp_type_t tstype;

		int64_t timestamp;

                rd_kafka_headers_t *hdrs;

		fprintf(stdout, "%% Message (offset %"PRId64", %zd bytes):\n",

			rkmessage->offset, rkmessage->len);

		timestamp = rd_kafka_message_timestamp(rkmessage, &tstype);

		if (tstype != RD_KAFKA_TIMESTAMP_NOT_AVAILABLE) {

			const char *tsname = "?";

			if (tstype == RD_KAFKA_TIMESTAMP_CREATE_TIME)

				tsname = "create time";

			else if (tstype == RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME)

				tsname = "log append time";

			fprintf(stdout, "%% Message timestamp: %s %"PRId64

				" (%ds ago)\n",

				tsname, timestamp,

				!timestamp ? 0 :

				(int)time(NULL) - (int)(timestamp/1000));

		}

                if (!rd_kafka_message_headers(rkmessage, &hdrs)) {

                        size_t idx = 0;

                        const char *name;

                        const void *val;

                        size_t size;

                        fprintf(stdout, "%% Headers:");

                        while (!rd_kafka_header_get_all(hdrs, idx++,

                                                        &name, &val, &size)) {

                                fprintf(stdout, "%s%s=",

                                        idx == 1 ? " " : ", ", name);

                                if (val)

                                        fprintf(stdout, "\"%.*s\"",

                                                (int)size, (const char *)val);

                                else

                                        fprintf(stdout, "NULL");

                        }

                        fprintf(stdout, "\n");

                }

	}

	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 sig_usr1 (int sig) {

	rd_kafka_dump(stdout, rk);

}

int main (int argc, char **argv) {

	rd_kafka_topic_t *rkt;

	char *brokers = "localhost:9092";

	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];

	int64_t start_offset = 0;

        int report_offsets = 0;

	int do_conf_dump = 0;

	char tmp[16];

        int64_t seek_offset = 0;

        int64_t tmp_offset = 0;

	int get_wmarks = 0;

        rd_kafka_headers_t *hdrs = NULL;

        rd_kafka_resp_err_t err;

	/* Kafka configuration */

	conf = rd_kafka_conf_new();

        /* Set logger */

        rd_kafka_conf_set_log_cb(conf, logger);

	/* Quick termination */

	snprintf(tmp, sizeof(tmp), "%i", SIGIO);

	rd_kafka_conf_set(conf, "internal.termination.signal", tmp, NULL, 0);

	/* Topic configuration */

	topic_conf = rd_kafka_topic_conf_new();

	while ((opt = getopt(argc, argv, "PCLt:p:b:z:qd:o:eX:As:H:")) != -1) {

		switch (opt) {

		case 'P':

		case 'C':

                case 'L':

			mode = opt;

			break;

		case 't':

			topic = optarg;

			break;

		case 'p':

			partition = atoi(optarg);

			break;

		case 'b':

			brokers = 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':

                case 's':

			if (!strcmp(optarg, "end"))

				tmp_offset = RD_KAFKA_OFFSET_END;

			else if (!strcmp(optarg, "beginning"))

				tmp_offset = RD_KAFKA_OFFSET_BEGINNING;

			else if (!strcmp(optarg, "stored"))

				tmp_offset = RD_KAFKA_OFFSET_STORED;

                        else if (!strcmp(optarg, "report"))

                                report_offsets = 1;

			else if (!strcmp(optarg, "wmark"))

				get_wmarks = 1;

			else {

				tmp_offset = strtoll(optarg, NULL, 10);

				if (tmp_offset < 0)

					tmp_offset = RD_KAFKA_OFFSET_TAIL(-tmp_offset);

			}

                        if (opt == 'o')

                                start_offset = tmp_offset;

                        else if (opt == 's')

                                seek_offset = tmp_offset;

			break;

		case 'e':

			exit_eof = 1;

			break;

		case 'd':

			if (rd_kafka_conf_set(conf, "debug", optarg,

					      errstr, sizeof(errstr)) !=

			    RD_KAFKA_CONF_OK) {

				fprintf(stderr,

					"%% Debug configuration failed: "

					"%s: %s\n",

					errstr, optarg);

				exit(1);

			}

			break;

		case 'q':

			quiet = 1;

			break;

		case 'A':

			output = OUTPUT_RAW;

			break;

                case 'H':

                {

                        char *name, *val;

                        size_t name_sz = -1;

                        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);

                        }

                }

                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, '='))) {

				char dest[512];

				size_t dest_size = sizeof(dest);

				/* Return current value for property. */

				res = RD_KAFKA_CONF_UNKNOWN;

				if (!strncmp(name, "topic.", strlen("topic.")))

					res = rd_kafka_topic_conf_get(

						topic_conf,

						name+strlen("topic."),

						dest, &dest_size);

				if (res == RD_KAFKA_CONF_UNKNOWN)

					res = rd_kafka_conf_get(

						conf, name, dest, &dest_size);

				if (res == RD_KAFKA_CONF_OK) {

					printf("%s = %s\n", name, dest);

					exit(0);

				} else {

					fprintf(stderr,

						"%% %s property\n",

						res == RD_KAFKA_CONF_UNKNOWN ?

						"Unknown" : "Invalid");

					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 < (int)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"

			"  -b <brokers>    Broker address (localhost:9092)\n"

			"  -z <codec>      Enable compression:\n"

			"                  none|gzip|snappy\n"

			"  -o <offset>     Start offset (consumer):\n"

			"                  beginning, end, NNNNN or -NNNNN\n"

			"                  wmark returns the current hi&lo "

			"watermarks.\n"

                        "  -o report       Report message offsets (producer)\n"

			"  -e              Exit consumer when last message\n"

			"                  in partition has been received.\n"

			"  -d [facs..]     Enable debugging contexts:\n"

			"                  %s\n"

			"  -q              Be quiet\n"

			"  -A              Raw payload output (consumer)\n"

                        "  -H <name[=value]> Add header to message (producer)\n"

			"  -X <prop=name>  Set arbitrary librdkafka "

			"configuration property\n"

			"                  Properties prefixed with \"topic.\" "

			"will be set on topic object.\n"

			"  -X list         Show full list of supported "

			"properties.\n"

			"  -X <prop>       Get single property value\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);

	}

	if ((mode == 'C' && !isatty(STDIN_FILENO)) ||

	    (mode != 'C' && !isatty(STDOUT_FILENO)))

		quiet = 1;

	signal(SIGINT, stop);

	signal(SIGUSR1, sig_usr1);

	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. */

                /* If offset reporting (-o report) is enabled, use the

                 * richer dr_msg_cb instead. */

                if (report_offsets) {

                        rd_kafka_topic_conf_set(topic_conf,

                                                "produce.offset.report",

                                                "true", errstr, sizeof(errstr));

                        rd_kafka_conf_set_dr_msg_cb(conf, msg_delivered2);

                } else

                        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);

		}

		/* Add brokers */

		if (rd_kafka_brokers_add(rk, brokers) == 0) {

			fprintf(stderr, "%% No valid brokers specified\n");

			exit(1);

		}

		/* Create topic */

		rkt = rd_kafka_topic_new(rk, topic, topic_conf);

                topic_conf = NULL; /* Now owned by topic */

		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 (hdrs) {

                                rd_kafka_headers_t *hdrs_copy;

                                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(RD_KAFKA_MSG_F_COPY),

                                        RD_KAFKA_V_VALUE(buf, len),

                                        RD_KAFKA_V_HEADERS(hdrs_copy),

                                        RD_KAFKA_V_END);

                                if (err)

                                        rd_kafka_headers_destroy(hdrs_copy);

                        } else {

                                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) {

                                        err = rd_kafka_last_error();

                                }

                        }

                        if (err) {

                                fprintf(stderr,

                                        "%% Failed to produce to topic %s "

					"partition %i: %s\n",

					rd_kafka_topic_name(rkt), partition,

					rd_kafka_err2str(err));

				/* 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 topic */

		rd_kafka_topic_destroy(rkt);

		/* 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);

		}

		/* Add brokers */

		if (rd_kafka_brokers_add(rk, brokers) == 0) {

			fprintf(stderr, "%% No valid brokers specified\n");

			exit(1);

		}

		if (get_wmarks) {

			int64_t lo, hi;

                        rd_kafka_resp_err_t err;

			/* Only query for hi&lo partition watermarks */

			if ((err = rd_kafka_query_watermark_offsets(

				     rk, topic, partition, &lo, &hi, 5000))) {

				fprintf(stderr, "%% query_watermark_offsets() "

					"failed: %s\n",

					rd_kafka_err2str(err));

				exit(1);

			}

			printf("%s [%d]: low - high offsets: "

			       "%"PRId64" - %"PRId64"\n",

			       topic, partition, lo, hi);

			rd_kafka_destroy(rk);

			exit(0);

		}

		/* Create topic */

		rkt = rd_kafka_topic_new(rk, topic, topic_conf);

                topic_conf = NULL; /* Now owned by topic */

		/* Start consuming */

		if (rd_kafka_consume_start(rkt, partition, start_offset) == -1){

			rd_kafka_resp_err_t err = rd_kafka_last_error();

			fprintf(stderr, "%% Failed to start consuming: %s\n",

				rd_kafka_err2str(err));

                        if (err == RD_KAFKA_RESP_ERR__INVALID_ARG)

                                fprintf(stderr,

                                        "%% Broker based offset storage "

                                        "requires a group.id, "

                                        "add: -X group.id=yourGroup\n");

			exit(1);

		}

		while (run) {

			rd_kafka_message_t *rkmessage;

                        rd_kafka_resp_err_t err;

                        /* Poll for errors, etc. */

                        rd_kafka_poll(rk, 0);

			/* 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);

                        if (seek_offset) {

                                err = rd_kafka_seek(rkt, partition, seek_offset,

                                                    2000);

                                if (err)

                                        printf("Seek failed: %s\n",

                                               rd_kafka_err2str(err));

                                else

                                        printf("Seeked to %"PRId64"\n",

                                               seek_offset);

                                seek_offset = 0;

                        }

		}

		/* Stop consuming */

		rd_kafka_consume_stop(rkt, partition);

                while (rd_kafka_outq_len(rk) > 0)

                        rd_kafka_poll(rk, 10);

		/* Destroy topic */

		rd_kafka_topic_destroy(rkt);

		/* Destroy handle */

		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);

		}

		/* Add brokers */

		if (rd_kafka_brokers_add(rk, brokers) == 0) {

			fprintf(stderr, "%% No valid brokers specified\n");

			exit(1);

		}

                /* Create topic */

                if (topic) {

                        rkt = rd_kafka_topic_new(rk, topic, topic_conf);

                        topic_conf = NULL; /* Now owned by topic */

                } 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 topic */

		if (rkt)

			rd_kafka_topic_destroy(rkt);

		/* Destroy the handle */

		rd_kafka_destroy(rk);

                if (topic_conf)

                        rd_kafka_topic_conf_destroy(topic_conf);

                /* Exit right away, dont wait for background cleanup, we haven't

                 * done anything important anyway. */

                exit(err ? 2 : 0);

        }

        if (hdrs)

                rd_kafka_headers_destroy(hdrs);

        if (topic_conf)

                rd_kafka_topic_conf_destroy(topic_conf);

	/* Let background threads clean up and terminate cleanly. */

	run = 5;

	while (run-- > 0 && rd_kafka_wait_destroyed(1000) == -1)

		printf("Waiting for librdkafka to decommission\n");