You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ph...@apache.org on 2018/06/06 14:14:36 UTC
[08/51] [partial] nifi-minifi-cpp git commit: MINIFICPP-512 - upgrade
to librdkafka 0.11.4
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.4/examples/CMakeLists.txt b/thirdparty/librdkafka-0.11.4/examples/CMakeLists.txt
new file mode 100644
index 0000000..dae7f9a
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.4/examples/CMakeLists.txt
@@ -0,0 +1,30 @@
+if(WIN32)
+ set(win32_sources ../win32/wingetopt.c ../win32/wingetopt.h)
+ set(win32_compile_defs "LIBRDKAFKACPP_EXPORTS=0")
+endif(WIN32)
+
+add_executable(rdkafka_simple_producer rdkafka_simple_producer.c ${win32_sources})
+target_link_libraries(rdkafka_simple_producer PUBLIC rdkafka)
+
+add_executable(rdkafka_performance rdkafka_performance.c ${win32_sources})
+target_link_libraries(rdkafka_performance PUBLIC rdkafka)
+
+add_executable(rdkafka_example_cpp rdkafka_example.cpp ${win32_sources})
+target_link_libraries(rdkafka_example_cpp PUBLIC rdkafka++)
+target_compile_definitions(rdkafka_example_cpp PRIVATE ${win32_compile_defs})
+
+add_executable(rdkafka_consumer_example_cpp rdkafka_consumer_example.cpp ${win32_sources})
+target_link_libraries(rdkafka_consumer_example_cpp PUBLIC rdkafka++)
+target_compile_definitions(rdkafka_consumer_example_cpp PRIVATE ${win32_compile_defs})
+
+# The targets below has Unix include dirs and do not compile on Windows.
+if(NOT WIN32)
+ add_executable(rdkafka_example rdkafka_example.c)
+ target_link_libraries(rdkafka_example PUBLIC rdkafka)
+
+ add_executable(rdkafka_consumer_example rdkafka_consumer_example.c)
+ target_link_libraries(rdkafka_consumer_example PUBLIC rdkafka)
+
+ add_executable(kafkatest_verifiable_client kafkatest_verifiable_client.cpp)
+ target_link_libraries(kafkatest_verifiable_client PUBLIC rdkafka++)
+endif(NOT WIN32)
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/Makefile
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.4/examples/Makefile b/thirdparty/librdkafka-0.11.4/examples/Makefile
new file mode 100644
index 0000000..d3e0832
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.4/examples/Makefile
@@ -0,0 +1,96 @@
+EXAMPLES ?= rdkafka_example rdkafka_performance rdkafka_example_cpp \
+ rdkafka_consumer_example rdkafka_consumer_example_cpp \
+ kafkatest_verifiable_client rdkafka_simple_producer
+
+all: $(EXAMPLES)
+
+include ../mklove/Makefile.base
+
+CFLAGS += -I../src
+CXXFLAGS += -I../src-cpp
+
+# librdkafka must be compiled with -gstrict-dwarf, but rdkafka_example must not,
+# due to some clang bug on OSX 10.9
+CPPFLAGS := $(subst strict-dwarf,,$(CPPFLAGS))
+
+rdkafka_example: ../src/librdkafka.a rdkafka_example.c
+ $(CC) $(CPPFLAGS) $(CFLAGS) rdkafka_example.c -o $@ $(LDFLAGS) \
+ ../src/librdkafka.a $(LIBS)
+ @echo "# $@ is ready"
+ @echo "#"
+ @echo "# Run producer (write messages on stdin)"
+ @echo "./$@ -P -t <topic> -p <partition>"
+ @echo ""
+ @echo "# or consumer"
+ @echo "./$@ -C -t <topic> -p <partition>"
+ @echo ""
+ @echo "#"
+ @echo "# More usage options:"
+ @echo "./$@ -h"
+
+rdkafka_simple_producer: ../src/librdkafka.a rdkafka_simple_producer.c
+ $(CC) $(CPPFLAGS) $(CFLAGS) $@.c -o $@ $(LDFLAGS) \
+ ../src/librdkafka.a $(LIBS)
+
+rdkafka_consumer_example: ../src/librdkafka.a rdkafka_consumer_example.c
+ $(CC) $(CPPFLAGS) $(CFLAGS) rdkafka_consumer_example.c -o $@ $(LDFLAGS) \
+ ../src/librdkafka.a $(LIBS)
+ @echo "# $@ is ready"
+ @echo "#"
+ @echo "./$@ <topic[:part]> <topic2[:part]> .."
+ @echo ""
+ @echo "#"
+ @echo "# More usage options:"
+ @echo "./$@ -h"
+
+rdkafka_performance: ../src/librdkafka.a rdkafka_performance.c
+ $(CC) $(CPPFLAGS) $(CFLAGS) rdkafka_performance.c -o $@ $(LDFLAGS) \
+ ../src/librdkafka.a $(LIBS)
+ @echo "# $@ is ready"
+ @echo "#"
+ @echo "# Run producer"
+ @echo "./$@ -P -t <topic> -p <partition> -s <msgsize>"
+ @echo ""
+ @echo "# or consumer"
+ @echo "./$@ -C -t <topic> -p <partition>"
+ @echo ""
+ @echo "#"
+ @echo "# More usage options:"
+ @echo "./$@ -h"
+
+
+rdkafka_example_cpp: ../src-cpp/librdkafka++.a ../src/librdkafka.a rdkafka_example.cpp
+ $(CXX) $(CPPFLAGS) $(CXXFLAGS) rdkafka_example.cpp -o $@ $(LDFLAGS) \
+ ../src-cpp/librdkafka++.a ../src/librdkafka.a $(LIBS) -lstdc++
+
+kafkatest_verifiable_client: ../src-cpp/librdkafka++.a ../src/librdkafka.a kafkatest_verifiable_client.cpp
+ $(CXX) $(CPPFLAGS) $(CXXFLAGS) kafkatest_verifiable_client.cpp -o $@ $(LDFLAGS) \
+ ../src-cpp/librdkafka++.a ../src/librdkafka.a $(LIBS) -lstdc++
+
+
+rdkafka_consumer_example_cpp: ../src-cpp/librdkafka++.a ../src/librdkafka.a rdkafka_consumer_example.cpp
+ $(CXX) $(CPPFLAGS) $(CXXFLAGS) rdkafka_consumer_example.cpp -o $@ $(LDFLAGS) \
+ ../src-cpp/librdkafka++.a ../src/librdkafka.a $(LIBS) -lstdc++
+
+rdkafka_consume_batch: ../src-cpp/librdkafka++.a ../src/librdkafka.a rdkafka_consume_batch.cpp
+ $(CXX) $(CPPFLAGS) $(CXXFLAGS) rdkafka_consume_batch.cpp -o $@ $(LDFLAGS) \
+ ../src-cpp/librdkafka++.a ../src/librdkafka.a $(LIBS) -lstdc++
+
+rdkafka_zookeeper_example: ../src/librdkafka.a rdkafka_zookeeper_example.c
+ $(CC) $(CPPFLAGS) $(CFLAGS) -I/usr/include/zookeeper rdkafka_zookeeper_example.c -o $@ $(LDFLAGS) \
+ ../src/librdkafka.a $(LIBS) -lzookeeper_mt -ljansson
+ @echo "# $@ is ready"
+ @echo "#"
+ @echo "# Run producer (write messages on stdin)"
+ @echo "./$@ -P -t <topic> -p <partition>"
+ @echo ""
+ @echo "# or consumer"
+ @echo "./$@ -C -t <topic> -p <partition>"
+ @echo ""
+ @echo "#"
+ @echo "# More usage options:"
+ @echo "./$@ -h"
+
+clean:
+ rm -f $(EXAMPLES)
+
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/globals.json
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.4/examples/globals.json b/thirdparty/librdkafka-0.11.4/examples/globals.json
new file mode 100644
index 0000000..527e126
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.4/examples/globals.json
@@ -0,0 +1,11 @@
+{"VerifiableConsumer":
+ {
+ "class": "kafkatest.services.verifiable_client.VerifiableClientApp",
+ "exec_cmd": "/vagrant/tests/c/kafkatest_verifiable_client --consumer --debug cgrp,topic,protocol,broker"
+ },
+ "VerifiableProducer":
+ {
+ "class": "kafkatest.services.verifiable_client.VerifiableClientApp",
+ "exec_cmd": "/vagrant/tests/c/kafkatest_verifiable_client --producer --debug topic,broker"
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/kafkatest_verifiable_client.cpp
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.4/examples/kafkatest_verifiable_client.cpp b/thirdparty/librdkafka-0.11.4/examples/kafkatest_verifiable_client.cpp
new file mode 100644
index 0000000..26e1ae0
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.4/examples/kafkatest_verifiable_client.cpp
@@ -0,0 +1,960 @@
+/*
+ * Copyright (c) 2015, Confluent Inc
+ * 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.
+ */
+
+/**
+ * librdkafka version of the Java VerifiableProducer and VerifiableConsumer
+ * for use with the official Kafka client tests.
+ */
+
+
+#include <iostream>
+#include <fstream>
+#include <sstream>
+#include <map>
+#include <string>
+#include <algorithm>
+#include <cstdlib>
+#include <cstdio>
+#include <csignal>
+#include <cstring>
+#include <unistd.h>
+#include <sys/time.h>
+#include <assert.h>
+#include <ctype.h>
+#include <strings.h>
+
+#ifdef _MSC_VER
+#include "../win32/wingetopt.h"
+#elif _AIX
+#include <unistd.h>
+#else
+#include <getopt.h>
+#endif
+
+/*
+ * Typically include path in a real application would be
+ * #include <librdkafka/rdkafkacpp.h>
+ */
+#include "rdkafkacpp.h"
+
+static bool run = true;
+static bool exit_eof = false;
+static int verbosity = 1;
+static std::string value_prefix;
+
+class Assignment {
+
+ public:
+ static std::string name (const std::string &t, int partition) {
+ std::stringstream stm;
+ stm << t << "." << partition;
+ return stm.str();
+ }
+
+ Assignment(): topic(""), partition(-1), consumedMessages(0),
+ minOffset(-1), maxOffset(0) {
+ printf("Created assignment\n");
+ }
+ Assignment(const Assignment &a) {
+ topic = a.topic;
+ partition = a.partition;
+ consumedMessages = a.consumedMessages;
+ minOffset = a.minOffset;
+ maxOffset = a.maxOffset;
+ }
+
+ Assignment &operator=(const Assignment &a) {
+ this->topic = a.topic;
+ this->partition = a.partition;
+ this->consumedMessages = a.consumedMessages;
+ this->minOffset = a.minOffset;
+ this->maxOffset = a.maxOffset;
+ return *this;
+ }
+
+ int operator==(const Assignment &a) const {
+ return !(this->topic == a.topic &&
+ this->partition == a.partition);
+ }
+
+ int operator<(const Assignment &a) const {
+ if (this->topic < a.topic) return 1;
+ if (this->topic >= a.topic) return 0;
+ return (this->partition < a.partition);
+ }
+
+ void setup (std::string t, int32_t p) {
+ assert(!t.empty());
+ assert(topic.empty() || topic == t);
+ assert(partition == -1 || partition == p);
+ topic = t;
+ partition = p;
+ }
+
+ std::string topic;
+ int partition;
+ int consumedMessages;
+ int64_t minOffset;
+ int64_t maxOffset;
+};
+
+
+
+
+static struct {
+ int maxMessages;
+
+ struct {
+ int numAcked;
+ int numSent;
+ int numErr;
+ } producer;
+
+ struct {
+ int consumedMessages;
+ int consumedMessagesLastReported;
+ int consumedMessagesAtLastCommit;
+ bool useAutoCommit;
+ std::map<std::string, Assignment> assignments;
+ } consumer;
+} state = {
+ /* .maxMessages = */ -1
+};
+
+
+static RdKafka::KafkaConsumer *consumer;
+
+
+static std::string now () {
+ struct timeval tv;
+ gettimeofday(&tv, NULL);
+ time_t t = tv.tv_sec;
+ struct tm tm;
+ char buf[64];
+
+ localtime_r(&t, &tm);
+ strftime(buf, sizeof(buf), "%H:%M:%S", &tm);
+ snprintf(buf+strlen(buf), sizeof(buf)-strlen(buf), ".%03d",
+ (int)(tv.tv_usec / 1000));
+
+ return buf;
+}
+
+
+static time_t watchdog_last_kick;
+static const int watchdog_timeout = 20; /* Must be > socket.timeout.ms */
+static void sigwatchdog (int sig) {
+ time_t t = time(NULL);
+ if (watchdog_last_kick + watchdog_timeout <= t) {
+ std::cerr << now() << ": WATCHDOG TIMEOUT (" <<
+ (int)(t - watchdog_last_kick) << "s): TERMINATING" << std::endl;
+ int *i = NULL;
+ *i = 100;
+ abort();
+ }
+}
+
+static void watchdog_kick () {
+ watchdog_last_kick = time(NULL);
+
+ /* Safe guard against hangs-on-exit */
+ alarm(watchdog_timeout);
+}
+
+
+
+
+
+static void errorString (const std::string &name,
+ const std::string &errmsg,
+ const std::string &topic,
+ const std::string *key,
+ const std::string &value) {
+ std::cout << "{ "
+ << "\"name\": \"" << name << "\", "
+ << "\"_time\": \"" << now() << "\", "
+ << "\"message\": \"" << errmsg << "\", "
+ << "\"topic\": \"" << topic << "\", "
+ << "\"key\": \"" << (key ? *key : "NULL") << "\", "
+ << "\"value\": \"" << value << "\" "
+ << "}" << std::endl;
+}
+
+
+static void successString (const std::string &name,
+ const std::string &topic,
+ int partition,
+ int64_t offset,
+ const std::string *key,
+ const std::string &value) {
+ std::cout << "{ "
+ << "\"name\": \"" << name << "\", "
+ << "\"_time\": \"" << now() << "\", "
+ << "\"topic\": \"" << topic << "\", "
+ << "\"partition\": " << partition << ", "
+ << "\"offset\": " << offset << ", "
+ << "\"key\": \"" << (key ? *key : "NULL") << "\", "
+ << "\"value\": \"" << value << "\" "
+ << "}" << std::endl;
+}
+
+
+#if FIXME
+static void offsetStatus (bool success,
+ const std::string &topic,
+ int partition,
+ int64_t offset,
+ const std::string &errstr) {
+ std::cout << "{ "
+ "\"name\": \"offsets_committed\", " <<
+ "\"success\": " << success << ", " <<
+ "\"offsets\": [ " <<
+ " { " <<
+ " \"topic\": \"" << topic << "\", " <<
+ " \"partition\": " << partition << ", " <<
+ " \"offset\": " << (int)offset << ", " <<
+ " \"error\": \"" << errstr << "\" " <<
+ " } " <<
+ "] }" << std::endl;
+
+}
+#endif
+
+
+static void sigterm (int sig) {
+
+ std::cerr << now() << ": Terminating because of signal " << sig << std::endl;
+
+ if (!run) {
+ std::cerr << now() << ": Forced termination" << std::endl;
+ exit(1);
+ }
+ run = false;
+}
+
+
+class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
+ public:
+ void dr_cb (RdKafka::Message &message) {
+ if (message.err()) {
+ state.producer.numErr++;
+ errorString("producer_send_error", message.errstr(),
+ message.topic_name(),
+ message.key(),
+ std::string(static_cast<const char*>(message.payload()),
+ message.len()));
+ } else {
+ successString("producer_send_success",
+ message.topic_name(),
+ (int)message.partition(),
+ message.offset(),
+ message.key(),
+ std::string(static_cast<const char*>(message.payload()),
+ message.len()));
+ state.producer.numAcked++;
+ }
+ }
+};
+
+
+class ExampleEventCb : public RdKafka::EventCb {
+ public:
+ void event_cb (RdKafka::Event &event) {
+ switch (event.type())
+ {
+ case RdKafka::Event::EVENT_ERROR:
+ std::cerr << now() << ": ERROR (" << RdKafka::err2str(event.err()) << "): " <<
+ event.str() << std::endl;
+ break;
+
+ case RdKafka::Event::EVENT_STATS:
+ std::cerr << now() << ": \"STATS\": " << event.str() << std::endl;
+ break;
+
+ case RdKafka::Event::EVENT_LOG:
+ std::cerr << now() << ": LOG-" << event.severity() << "-"
+ << event.fac() << ": " << event.str() << std::endl;
+ break;
+
+ default:
+ std::cerr << now() << ": EVENT " << event.type() <<
+ " (" << RdKafka::err2str(event.err()) << "): " <<
+ event.str() << std::endl;
+ break;
+ }
+ }
+};
+
+
+/* Use of this partitioner is pretty pointless since no key is provided
+ * in the produce() call. */
+class MyHashPartitionerCb : public RdKafka::PartitionerCb {
+ public:
+ int32_t partitioner_cb (const RdKafka::Topic *topic, const std::string *key,
+ int32_t partition_cnt, void *msg_opaque) {
+ return djb_hash(key->c_str(), key->size()) % partition_cnt;
+ }
+ private:
+
+ static inline unsigned int djb_hash (const char *str, size_t len) {
+ unsigned int hash = 5381;
+ for (size_t i = 0 ; i < len ; i++)
+ hash = ((hash << 5) + hash) + str[i];
+ return hash;
+ }
+};
+
+
+
+
+
+/**
+ * Print number of records consumed, every 100 messages or on timeout.
+ */
+static void report_records_consumed (int immediate) {
+ std::map<std::string,Assignment> *assignments = &state.consumer.assignments;
+
+ if (state.consumer.consumedMessages <=
+ state.consumer.consumedMessagesLastReported + (immediate ? 0 : 999))
+ return;
+
+ std::cout << "{ "
+ "\"name\": \"records_consumed\", " <<
+ "\"_totcount\": " << state.consumer.consumedMessages << ", " <<
+ "\"count\": " << (state.consumer.consumedMessages -
+ state.consumer.consumedMessagesLastReported) << ", " <<
+ "\"partitions\": [ ";
+
+ for (std::map<std::string,Assignment>::iterator ii = assignments->begin() ;
+ ii != assignments->end() ; ii++) {
+ Assignment *a = &(*ii).second;
+ assert(!a->topic.empty());
+ std::cout << (ii == assignments->begin() ? "": ", ") << " { " <<
+ " \"topic\": \"" << a->topic << "\", " <<
+ " \"partition\": " << a->partition << ", " <<
+ " \"minOffset\": " << a->minOffset << ", " <<
+ " \"maxOffset\": " << a->maxOffset << " " <<
+ " } ";
+ a->minOffset = -1;
+ }
+
+ std::cout << "] }" << std::endl;
+
+ state.consumer.consumedMessagesLastReported = state.consumer.consumedMessages;
+}
+
+
+class ExampleOffsetCommitCb : public RdKafka::OffsetCommitCb {
+ public:
+ void offset_commit_cb (RdKafka::ErrorCode err,
+ std::vector<RdKafka::TopicPartition*> &offsets) {
+ std::cerr << now() << ": Propagate offset for " << offsets.size() << " partitions, error: " << RdKafka::err2str(err) << std::endl;
+
+ /* No offsets to commit, dont report anything. */
+ if (err == RdKafka::ERR__NO_OFFSET)
+ return;
+
+ /* Send up-to-date records_consumed report to make sure consumed > committed */
+ report_records_consumed(1);
+
+ std::cout << "{ " <<
+ "\"name\": \"offsets_committed\", " <<
+ "\"success\": " << (err ? "false" : "true") << ", " <<
+ "\"error\": \"" << (err ? RdKafka::err2str(err) : "") << "\", " <<
+ "\"_autocommit\": " << (state.consumer.useAutoCommit ? "true":"false") << ", " <<
+ "\"offsets\": [ ";
+ assert(offsets.size() > 0);
+ for (unsigned int i = 0 ; i < offsets.size() ; i++) {
+ std::cout << (i == 0 ? "" : ", ") << "{ " <<
+ " \"topic\": \"" << offsets[i]->topic() << "\", " <<
+ " \"partition\": " << offsets[i]->partition() << ", " <<
+ " \"offset\": " << (int)offsets[i]->offset() << ", " <<
+ " \"error\": \"" <<
+ (offsets[i]->err() ? RdKafka::err2str(offsets[i]->err()) : "") <<
+ "\" " <<
+ " }";
+ }
+ std::cout << " ] }" << std::endl;
+
+ }
+};
+
+static ExampleOffsetCommitCb ex_offset_commit_cb;
+
+
+/**
+ * Commit every 1000 messages or whenever there is a consume timeout.
+ */
+static void do_commit (RdKafka::KafkaConsumer *consumer,
+ int immediate) {
+ if (!immediate &&
+ (state.consumer.useAutoCommit ||
+ state.consumer.consumedMessagesAtLastCommit + 1000 >
+ state.consumer.consumedMessages))
+ return;
+
+ /* Make sure we report consumption before commit,
+ * otherwise tests may fail because of commit > consumed. */
+ if (state.consumer.consumedMessagesLastReported <
+ state.consumer.consumedMessages)
+ report_records_consumed(1);
+
+ std::cerr << now() << ": committing " <<
+ (state.consumer.consumedMessages -
+ state.consumer.consumedMessagesAtLastCommit) << " messages" << std::endl;
+
+ RdKafka::ErrorCode err;
+ err = consumer->commitSync(&ex_offset_commit_cb);
+
+ std::cerr << now() << ": " <<
+ "sync commit returned " << RdKafka::err2str(err) << std::endl;
+
+ state.consumer.consumedMessagesAtLastCommit =
+ state.consumer.consumedMessages;
+}
+
+
+void msg_consume(RdKafka::KafkaConsumer *consumer,
+ RdKafka::Message* msg, void* opaque) {
+ switch (msg->err()) {
+ case RdKafka::ERR__TIMED_OUT:
+ /* Try reporting consumed messages */
+ report_records_consumed(1);
+ /* Commit one every consume() timeout instead of on every message.
+ * Also commit on every 1000 messages, whichever comes first. */
+ do_commit(consumer, 1);
+ break;
+
+
+ case RdKafka::ERR_NO_ERROR:
+ {
+ /* Real message */
+ if (verbosity > 2)
+ std::cerr << now() << ": Read msg from " << msg->topic_name() <<
+ " [" << (int)msg->partition() << "] at offset " <<
+ msg->offset() << std::endl;
+
+ if (state.maxMessages >= 0 &&
+ state.consumer.consumedMessages >= state.maxMessages)
+ return;
+
+
+ Assignment *a =
+ &state.consumer.assignments[Assignment::name(msg->topic_name(),
+ msg->partition())];
+ a->setup(msg->topic_name(), msg->partition());
+
+ a->consumedMessages++;
+ if (a->minOffset == -1)
+ a->minOffset = msg->offset();
+ if (a->maxOffset < msg->offset())
+ a->maxOffset = msg->offset();
+
+ if (msg->key()) {
+ if (verbosity >= 3)
+ std::cerr << now() << ": Key: " << *msg->key() << std::endl;
+ }
+
+ if (verbosity >= 3)
+ fprintf(stderr, "%.*s\n",
+ static_cast<int>(msg->len()),
+ static_cast<const char *>(msg->payload()));
+
+ state.consumer.consumedMessages++;
+
+ report_records_consumed(0);
+
+ do_commit(consumer, 0);
+ }
+ break;
+
+ case RdKafka::ERR__PARTITION_EOF:
+ /* Last message */
+ if (exit_eof) {
+ std::cerr << now() << ": Terminate: exit on EOF" << std::endl;
+ run = false;
+ }
+ break;
+
+ case RdKafka::ERR__UNKNOWN_TOPIC:
+ case RdKafka::ERR__UNKNOWN_PARTITION:
+ std::cerr << now() << ": Consume failed: " << msg->errstr() << std::endl;
+ run = false;
+ break;
+
+ case RdKafka::ERR_GROUP_COORDINATOR_NOT_AVAILABLE:
+ std::cerr << now() << ": Warning: " << msg->errstr() << std::endl;
+ break;
+
+ default:
+ /* Errors */
+ std::cerr << now() << ": Consume failed: " << msg->errstr() << std::endl;
+ run = false;
+ }
+}
+
+
+
+
+class ExampleConsumeCb : public RdKafka::ConsumeCb {
+ public:
+ void consume_cb (RdKafka::Message &msg, void *opaque) {
+ msg_consume(consumer_, &msg, opaque);
+ }
+ RdKafka::KafkaConsumer *consumer_;
+};
+
+class ExampleRebalanceCb : public RdKafka::RebalanceCb {
+ private:
+ static std::string part_list_json (const std::vector<RdKafka::TopicPartition*> &partitions) {
+ std::ostringstream out;
+ for (unsigned int i = 0 ; i < partitions.size() ; i++)
+ out << (i==0?"":", ") << "{ " <<
+ " \"topic\": \"" << partitions[i]->topic() << "\", " <<
+ " \"partition\": " << partitions[i]->partition() <<
+ " }";
+ return out.str();
+ }
+ public:
+ void rebalance_cb (RdKafka::KafkaConsumer *consumer,
+ RdKafka::ErrorCode err,
+ std::vector<RdKafka::TopicPartition*> &partitions) {
+
+ std::cerr << now() << ": rebalance_cb " << RdKafka::err2str(err) <<
+ " for " << partitions.size() << " partitions" << std::endl;
+ /* Send message report prior to rebalancing event to make sure they
+ * are accounted for on the "right side" of the rebalance. */
+ report_records_consumed(1);
+
+ if (err == RdKafka::ERR__ASSIGN_PARTITIONS)
+ consumer->assign(partitions);
+ else {
+ do_commit(consumer, 1);
+ consumer->unassign();
+ }
+
+ std::cout <<
+ "{ " <<
+ "\"name\": \"partitions_" << (err == RdKafka::ERR__ASSIGN_PARTITIONS ?
+ "assigned" : "revoked") << "\", " <<
+ "\"partitions\": [ " << part_list_json(partitions) << "] }" << std::endl;
+
+ }
+};
+
+
+
+/**
+ * @brief Read (Java client) configuration file
+ */
+static void read_conf_file (RdKafka::Conf *conf, const std::string &conf_file) {
+ std::ifstream inf(conf_file.c_str());
+
+ if (!inf) {
+ std::cerr << now() << ": " << conf_file << ": could not open file" << std::endl;
+ exit(1);
+ }
+
+ std::cerr << now() << ": " << conf_file << ": read config file" << std::endl;
+
+ std::string line;
+ int linenr = 0;
+
+ while (std::getline(inf, line)) {
+ linenr++;
+
+ // Ignore comments and empty lines
+ if (line[0] == '#' || line.length() == 0)
+ continue;
+
+ // Match on key=value..
+ size_t d = line.find("=");
+ if (d == 0 || d == std::string::npos) {
+ std::cerr << now() << ": " << conf_file << ":" << linenr << ": " << line << ": ignoring invalid line (expect key=value): " << ::std::endl;
+ continue;
+ }
+
+ std::string key = line.substr(0, d);
+ std::string val = line.substr(d+1);
+
+ std::string errstr;
+ if (conf->set(key, val, errstr)) {
+ std::cerr << now() << ": " << conf_file << ":" << linenr << ": " << key << "=" << val << ": " << errstr << ": ignoring error" << std::endl;
+ } else {
+ std::cerr << now() << ": " << conf_file << ":" << linenr << ": " << key << "=" << val << ": applied to configuration" << std::endl;
+ }
+ }
+
+ inf.close();
+}
+
+
+
+
+int main (int argc, char **argv) {
+ std::string brokers = "localhost";
+ std::string errstr;
+ std::vector<std::string> topics;
+ std::string mode = "P";
+ int throughput = 0;
+ int32_t partition = RdKafka::Topic::PARTITION_UA;
+ MyHashPartitionerCb hash_partitioner;
+ int64_t create_time = -1;
+
+ std::cerr << now() << ": librdkafka version " << RdKafka::version_str() <<
+ " (" << RdKafka::version() << ")" << std::endl;
+
+ /*
+ * Create configuration objects
+ */
+ RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
+
+ /* Java VerifiableProducer defaults to acks=all */
+ if (conf->set("acks", "all", errstr)) {
+ std::cerr << now() << ": " << errstr << std::endl;
+ exit(1);
+ }
+
+ /* Avoid slow shutdown on error */
+ if (conf->set("message.timeout.ms", "60000", errstr)) {
+ std::cerr << now() << ": " << errstr << std::endl;
+ exit(1);
+ }
+
+ {
+ char hostname[128];
+ gethostname(hostname, sizeof(hostname)-1);
+ conf->set("client.id", std::string("rdkafka@") + hostname, errstr);
+ }
+
+ conf->set("log.thread.name", "true", errstr);
+
+ /* correct producer offsets */
+ conf->set("produce.offset.report", "true", errstr);
+
+ /* auto commit is explicitly enabled with --enable-autocommit */
+ conf->set("enable.auto.commit", "false", errstr);
+
+ /* keep protocol request timeouts under the watchdog timeout
+ * to make sure things like commitSync() dont fall victim to the watchdog. */
+ conf->set("socket.timeout.ms", "10000", errstr);
+
+ conf->set("fetch.wait.max.ms", "500", errstr);
+ conf->set("fetch.min.bytes", "4096", errstr);
+
+ for (int i = 1 ; i < argc ; i++) {
+ const char *name = argv[i];
+ const char *val = i+1 < argc ? argv[i+1] : NULL;
+
+ if (val && !strncmp(val, "-", 1))
+ val = NULL;
+
+ std::cout << now() << ": argument: " << name << " " <<
+ (val?val:"") << std::endl;
+
+ if (val) {
+ if (!strcmp(name, "--topic"))
+ topics.push_back(val);
+ else if (!strcmp(name, "--broker-list"))
+ brokers = val;
+ else if (!strcmp(name, "--max-messages"))
+ state.maxMessages = atoi(val);
+ else if (!strcmp(name, "--throughput"))
+ throughput = atoi(val);
+ else if (!strcmp(name, "--producer.config") ||
+ !strcmp(name, "--consumer.config"))
+ read_conf_file(conf, val);
+ else if (!strcmp(name, "--group-id"))
+ conf->set("group.id", val, errstr);
+ else if (!strcmp(name, "--session-timeout"))
+ conf->set("session.timeout.ms", val, errstr);
+ else if (!strcmp(name, "--reset-policy")) {
+ if (conf->set("auto.offset.reset", val, errstr)) {
+ std::cerr << now() << ": " << errstr << std::endl;
+ exit(1);
+ }
+ } else if (!strcmp(name, "--assignment-strategy")) {
+ /* The system tests pass the Java class name(s) rather than
+ * the configuration value. Fix it.
+ * "org.apache.kafka.clients.consumer.RangeAssignor,.." -> "range,.."
+ */
+ std::string s = val;
+ size_t pos;
+
+ while ((pos = s.find("org.apache.kafka.clients.consumer.")) !=
+ std::string::npos)
+ s.erase(pos, strlen("org.apache.kafka.clients.consumer."));
+
+ while ((pos = s.find("Assignor")) != std::string::npos)
+ s.erase(pos, strlen("Assignor"));
+
+ std::transform(s.begin(), s.end(), s.begin(), tolower);
+
+ std::cerr << now() << ": converted " << name << " "
+ << val << " to " << s << std::endl;
+
+ if (conf->set("partition.assignment.strategy", s.c_str(), errstr)) {
+ std::cerr << now() << ": " << errstr << std::endl;
+ exit(1);
+ }
+ } else if (!strcmp(name, "--value-prefix")) {
+ value_prefix = std::string(val) + ".";
+ } else if (!strcmp(name, "--acks")) {
+ if (conf->set("acks", val, errstr)) {
+ std::cerr << now() << ": " << errstr << std::endl;
+ exit(1);
+ }
+ } else if (!strcmp(name, "--message-create-time")) {
+ create_time = (int64_t)atoi(val);
+ } else if (!strcmp(name, "--debug")) {
+ conf->set("debug", val, errstr);
+ } else if (!strcmp(name, "-X")) {
+ char *s = strdup(val);
+ char *t = strchr(s, '=');
+ if (!t)
+ t = (char *)"";
+ else {
+ *t = '\0';
+ t++;
+ }
+ if (conf->set(s, t, errstr)) {
+ std::cerr << now() << ": " << errstr << std::endl;
+ exit(1);
+ }
+ free(s);
+ } else {
+ std::cerr << now() << ": Unknown option " << name << std::endl;
+ exit(1);
+ }
+
+ i++;
+
+ } else {
+ if (!strcmp(name, "--consumer"))
+ mode = "C";
+ else if (!strcmp(name, "--producer"))
+ mode = "P";
+ else if (!strcmp(name, "--enable-autocommit")) {
+ state.consumer.useAutoCommit = true;
+ conf->set("enable.auto.commit", "true", errstr);
+ } else if (!strcmp(name, "-v"))
+ verbosity++;
+ else if (!strcmp(name, "-q"))
+ verbosity--;
+ else {
+ std::cerr << now() << ": Unknown option or missing argument to " << name << std::endl;
+ exit(1);
+ }
+ }
+ }
+
+ if (topics.empty() || brokers.empty()) {
+ std::cerr << now() << ": Missing --topic and --broker-list" << std::endl;
+ exit(1);
+ }
+
+
+ /*
+ * Set configuration properties
+ */
+ conf->set("metadata.broker.list", brokers, errstr);
+
+ ExampleEventCb ex_event_cb;
+ conf->set("event_cb", &ex_event_cb, errstr);
+
+ signal(SIGINT, sigterm);
+ signal(SIGTERM, sigterm);
+ signal(SIGALRM, sigwatchdog);
+
+
+ if (mode == "P") {
+ /*
+ * Producer mode
+ */
+
+ ExampleDeliveryReportCb ex_dr_cb;
+
+ /* Set delivery report callback */
+ conf->set("dr_cb", &ex_dr_cb, errstr);
+
+ /*
+ * Create producer using accumulated global configuration.
+ */
+ RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
+ if (!producer) {
+ std::cerr << now() << ": Failed to create producer: " << errstr << std::endl;
+ exit(1);
+ }
+
+ std::cerr << now() << ": % Created producer " << producer->name() << std::endl;
+
+ /*
+ * Create topic handle.
+ */
+ RdKafka::Topic *topic = RdKafka::Topic::create(producer, topics[0],
+ NULL, errstr);
+ if (!topic) {
+ std::cerr << now() << ": Failed to create topic: " << errstr << std::endl;
+ exit(1);
+ }
+
+ static const int delay_us = throughput ? 1000000/throughput : 10;
+
+ if (state.maxMessages == -1)
+ state.maxMessages = 1000000; /* Avoid infinite produce */
+
+ for (int i = 0 ; run && i < state.maxMessages ; i++) {
+ /*
+ * Produce message
+ */
+ std::ostringstream msg;
+ msg << value_prefix << i;
+ while (true) {
+ RdKafka::ErrorCode resp;
+ if (create_time == -1) {
+ resp = producer->produce(topic, partition,
+ RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
+ const_cast<char *>(msg.str().c_str()),
+ msg.str().size(), NULL, NULL);
+ } else {
+ resp = producer->produce(topics[0], partition,
+ RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
+ const_cast<char *>(msg.str().c_str()),
+ msg.str().size(),
+ NULL, 0,
+ create_time,
+ NULL);
+ }
+
+ if (resp == RdKafka::ERR__QUEUE_FULL) {
+ producer->poll(100);
+ continue;
+ } else if (resp != RdKafka::ERR_NO_ERROR) {
+ errorString("producer_send_error",
+ RdKafka::err2str(resp), topic->name(), NULL, msg.str());
+ state.producer.numErr++;
+ } else {
+ state.producer.numSent++;
+ }
+ break;
+ }
+
+ producer->poll(delay_us / 1000);
+ usleep(1000);
+ watchdog_kick();
+ }
+ run = true;
+
+ while (run && producer->outq_len() > 0) {
+ std::cerr << now() << ": Waiting for " << producer->outq_len() << std::endl;
+ producer->poll(1000);
+ watchdog_kick();
+ }
+
+ std::cerr << now() << ": " << state.producer.numAcked << "/" <<
+ state.producer.numSent << "/" << state.maxMessages <<
+ " msgs acked/sent/max, " << state.producer.numErr <<
+ " errored" << std::endl;
+
+ delete topic;
+ delete producer;
+
+
+ } else if (mode == "C") {
+ /*
+ * Consumer mode
+ */
+
+ conf->set("auto.offset.reset", "smallest", errstr);
+
+ ExampleRebalanceCb ex_rebalance_cb;
+ conf->set("rebalance_cb", &ex_rebalance_cb, errstr);
+
+ conf->set("offset_commit_cb", &ex_offset_commit_cb, errstr);
+
+
+ /*
+ * Create consumer using accumulated global configuration.
+ */
+ consumer = RdKafka::KafkaConsumer::create(conf, errstr);
+ if (!consumer) {
+ std::cerr << now() << ": Failed to create consumer: " <<
+ errstr << std::endl;
+ exit(1);
+ }
+
+ std::cerr << now() << ": % Created consumer " << consumer->name() <<
+ std::endl;
+
+ /*
+ * Subscribe to topic(s)
+ */
+ RdKafka::ErrorCode resp = consumer->subscribe(topics);
+ if (resp != RdKafka::ERR_NO_ERROR) {
+ std::cerr << now() << ": Failed to subscribe to " << topics.size() << " topics: "
+ << RdKafka::err2str(resp) << std::endl;
+ exit(1);
+ }
+
+ watchdog_kick();
+
+ /*
+ * Consume messages
+ */
+ while (run) {
+ RdKafka::Message *msg = consumer->consume(500);
+ msg_consume(consumer, msg, NULL);
+ delete msg;
+ watchdog_kick();
+ }
+
+ std::cerr << now() << ": Final commit on termination" << std::endl;
+
+ /* Final commit */
+ do_commit(consumer, 1);
+
+ /*
+ * Stop consumer
+ */
+ consumer->close();
+
+ delete consumer;
+ }
+
+ std::cout << "{ \"name\": \"shutdown_complete\" }" << std::endl;
+
+ /*
+ * Wait for RdKafka to decommission.
+ * This is not strictly needed (when check outq_len() above), but
+ * allows RdKafka to clean up all its resources before the application
+ * exits so that memory profilers such as valgrind wont complain about
+ * memory leaks.
+ */
+ RdKafka::wait_destroyed(5000);
+
+ std::cerr << now() << ": EXITING WITH RETURN VALUE 0" << std::endl;
+ return 0;
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/rdkafka_consume_batch.cpp
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.4/examples/rdkafka_consume_batch.cpp b/thirdparty/librdkafka-0.11.4/examples/rdkafka_consume_batch.cpp
new file mode 100644
index 0000000..ea4a169
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.4/examples/rdkafka_consume_batch.cpp
@@ -0,0 +1,260 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2018, 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)
+ *
+ * This example shows how to read batches of messages.
+ * Note that messages are fetched from the broker in batches regardless
+ * of how the application polls messages from librdkafka, this example
+ * merely shows how to accumulate a set of messages in the application.
+ */
+
+#include <iostream>
+#include <string>
+#include <cstdlib>
+#include <cstdio>
+#include <csignal>
+#include <cstring>
+
+#ifndef _MSC_VER
+#include <sys/time.h>
+#endif
+
+#ifdef _MSC_VER
+#include "../win32/wingetopt.h"
+#include <atltime.h>
+#elif _AIX
+#include <unistd.h>
+#else
+#include <getopt.h>
+#include <unistd.h>
+#endif
+
+/*
+ * Typically include path in a real application would be
+ * #include <librdkafka/rdkafkacpp.h>
+ */
+#include "rdkafkacpp.h"
+
+
+
+static bool run = true;
+
+static void sigterm (int sig) {
+ run = false;
+}
+
+
+
+/**
+ * @returns the current wall-clock time in milliseconds
+ */
+static int64_t now () {
+#ifndef _MSC_VER
+ struct timeval tv;
+ gettimeofday(&tv, NULL);
+ return ((int64_t)tv.tv_sec * 1000) + (tv.tv_usec / 1000);
+#else
+#error "now() not implemented for Windows, please submit a PR"
+#endif
+}
+
+
+
+/**
+ * @brief Accumulate a batch of \p batch_size messages, but wait
+ * no longer than \p batch_tmout milliseconds.
+ */
+static std::vector<RdKafka::Message *>
+consume_batch (RdKafka::KafkaConsumer *consumer, size_t batch_size, int batch_tmout) {
+
+ std::vector<RdKafka::Message *> msgs;
+ msgs.reserve(batch_size);
+
+ int64_t end = now() + batch_tmout;
+ int remaining_timeout = batch_tmout;
+
+ while (msgs.size() < batch_size) {
+ RdKafka::Message *msg = consumer->consume(remaining_timeout);
+
+ switch (msg->err()) {
+ case RdKafka::ERR__TIMED_OUT:
+ delete msg;
+ return msgs;
+
+ case RdKafka::ERR_NO_ERROR:
+ msgs.push_back(msg);
+ break;
+
+ default:
+ std::cerr << "%% Consumer error: " << msg->errstr() << std::endl;
+ run = false;
+ delete msg;
+ return msgs;
+ }
+
+ remaining_timeout = end - now();
+ if (remaining_timeout < 0)
+ break;
+ }
+
+ return msgs;
+}
+
+
+int main (int argc, char **argv) {
+ std::string errstr;
+ std::string topic_str;
+ std::vector<std::string> topics;
+ int batch_size = 100;
+ int batch_tmout = 1000;
+
+ /* Create configuration objects */
+ RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
+
+ if (conf->set("enable.partition.eof", "false", errstr) != RdKafka::Conf::CONF_OK) {
+ std::cerr << errstr << std::endl;
+ exit(1);
+ }
+
+ /* Read command line arguments */
+ int opt;
+ while ((opt = getopt(argc, argv, "g:B:T::b:X:")) != -1) {
+ switch (opt) {
+ case 'g':
+ if (conf->set("group.id", optarg, errstr) != RdKafka::Conf::CONF_OK) {
+ std::cerr << errstr << std::endl;
+ exit(1);
+ }
+ break;
+
+ case 'B':
+ batch_size = atoi(optarg);
+ break;
+
+ case 'T':
+ batch_tmout = atoi(optarg);
+ break;
+
+ case 'b':
+ if (conf->set("bootstrap.servers", optarg, errstr) != RdKafka::Conf::CONF_OK) {
+ std::cerr << errstr << std::endl;
+ exit(1);
+ }
+ break;
+
+ case 'X':
+ {
+ char *name, *val;
+
+ name = optarg;
+ if (!(val = strchr(name, '='))) {
+ std::cerr << "%% Expected -X property=value, not " <<
+ name << std::endl;
+ exit(1);
+ }
+
+ *val = '\0';
+ val++;
+
+ if (conf->set(name, val, errstr) != RdKafka::Conf::CONF_OK) {
+ std::cerr << errstr << std::endl;
+ exit(1);
+ }
+ }
+ break;
+
+ default:
+ goto usage;
+ }
+ }
+
+ /* Topics to consume */
+ for (; optind < argc ; optind++)
+ topics.push_back(std::string(argv[optind]));
+
+ if (topics.empty() || optind != argc) {
+ usage:
+ fprintf(stderr,
+ "Usage: %s -g <group-id> -B <batch-size> [options] topic1 topic2..\n"
+ "\n"
+ "librdkafka version %s (0x%08x)\n"
+ "\n"
+ " Options:\n"
+ " -g <group-id> Consumer group id\n"
+ " -B <batch-size> How many messages to batch (default: 100).\n"
+ " -T <batch-tmout> How long to wait for batch-size to accumulate in milliseconds. (default 1000 ms)\n"
+ " -b <brokers> Broker address (localhost:9092)\n"
+ " -X <prop=name> Set arbitrary librdkafka configuration property\n"
+ "\n",
+ argv[0],
+ RdKafka::version_str().c_str(), RdKafka::version());
+ exit(1);
+ }
+
+
+ signal(SIGINT, sigterm);
+ signal(SIGTERM, sigterm);
+
+ /* Create consumer */
+ RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);
+ if (!consumer) {
+ std::cerr << "Failed to create consumer: " << errstr << std::endl;
+ exit(1);
+ }
+
+ delete conf;
+
+ /* Subscribe to topics */
+ RdKafka::ErrorCode err = consumer->subscribe(topics);
+ if (err) {
+ std::cerr << "Failed to subscribe to " << topics.size() << " topics: "
+ << RdKafka::err2str(err) << std::endl;
+ exit(1);
+ }
+
+ /* Consume messages in batches of \p batch_size */
+ while (run) {
+ auto msgs = consume_batch(consumer, batch_size, batch_tmout);
+ std::cout << "Accumulated " << msgs.size() << " messages:" << std::endl;
+
+ for (auto &msg : msgs) {
+ std::cout << " Message in " << msg->topic_name() << " [" << msg->partition() << "] at offset " << msg->offset() << std::endl;
+ delete msg;
+ }
+ }
+
+ /* Close and destroy consumer */
+ consumer->close();
+ delete consumer;
+
+ return 0;
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/rdkafka_consumer_example.c
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.4/examples/rdkafka_consumer_example.c b/thirdparty/librdkafka-0.11.4/examples/rdkafka_consumer_example.c
new file mode 100644
index 0000000..3896df8
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.4/examples/rdkafka_consumer_example.c
@@ -0,0 +1,624 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2015, Magnus Edenhill
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ * this list of conditions and the following disclaimer in the documentation
+ * and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+/**
+ * Apache Kafka high level consumer example program
+ * 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>
+#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 wait_eof = 0; /* number of partitions awaiting EOF */
+static int quiet = 0;
+static enum {
+ OUTPUT_HEXDUMP,
+ OUTPUT_RAW,
+} output = OUTPUT_HEXDUMP;
+
+static void stop (int sig) {
+ if (!run)
+ exit(1);
+ 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;
+ unsigned 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 < (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, "%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(stdout, "%u.%03u RDKAFKA-%i-%s: %s: %s\n",
+ (int)tv.tv_sec, (int)(tv.tv_usec / 1000),
+ level, fac, rd_kafka_name(rk), buf);
+}
+
+
+
+/**
+ * Handle and print a consumed message.
+ * Internally crafted messages are also used to propagate state from
+ * librdkafka to the application. The application needs to check
+ * the `rkmessage->err` field for this purpose.
+ */
+static void msg_consume (rd_kafka_message_t *rkmessage) {
+ 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 && --wait_eof == 0) {
+ fprintf(stderr,
+ "%% All partition(s) reached EOF: "
+ "exiting\n");
+ run = 0;
+ }
+
+ return;
+ }
+
+ if (rkmessage->rkt)
+ 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));
+ else
+ fprintf(stderr, "%% Consumer error: %s: %s\n",
+ rd_kafka_err2str(rkmessage->err),
+ 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)
+ fprintf(stdout, "%% Message (topic %s [%"PRId32"], "
+ "offset %"PRId64", %zd bytes):\n",
+ rd_kafka_topic_name(rkmessage->rkt),
+ rkmessage->partition,
+ 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 print_partition_list (FILE *fp,
+ const rd_kafka_topic_partition_list_t
+ *partitions) {
+ int i;
+ for (i = 0 ; i < partitions->cnt ; i++) {
+ fprintf(stderr, "%s %s [%"PRId32"] offset %"PRId64,
+ i > 0 ? ",":"",
+ partitions->elems[i].topic,
+ partitions->elems[i].partition,
+ partitions->elems[i].offset);
+ }
+ fprintf(stderr, "\n");
+
+}
+static void rebalance_cb (rd_kafka_t *rk,
+ rd_kafka_resp_err_t err,
+ rd_kafka_topic_partition_list_t *partitions,
+ void *opaque) {
+
+ fprintf(stderr, "%% Consumer group rebalanced: ");
+
+ switch (err)
+ {
+ case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
+ fprintf(stderr, "assigned:\n");
+ print_partition_list(stderr, partitions);
+ rd_kafka_assign(rk, partitions);
+ wait_eof += partitions->cnt;
+ break;
+
+ case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
+ fprintf(stderr, "revoked:\n");
+ print_partition_list(stderr, partitions);
+ rd_kafka_assign(rk, NULL);
+ wait_eof = 0;
+ break;
+
+ default:
+ fprintf(stderr, "failed: %s\n",
+ rd_kafka_err2str(err));
+ rd_kafka_assign(rk, NULL);
+ break;
+ }
+}
+
+
+static int describe_groups (rd_kafka_t *rk, const char *group) {
+ rd_kafka_resp_err_t err;
+ const struct rd_kafka_group_list *grplist;
+ int i;
+
+ err = rd_kafka_list_groups(rk, group, &grplist, 10000);
+
+ if (err) {
+ fprintf(stderr, "%% Failed to acquire group list: %s\n",
+ rd_kafka_err2str(err));
+ return -1;
+ }
+
+ for (i = 0 ; i < grplist->group_cnt ; i++) {
+ const struct rd_kafka_group_info *gi = &grplist->groups[i];
+ int j;
+
+ printf("Group \"%s\" in state %s on broker %d (%s:%d)\n",
+ gi->group, gi->state,
+ gi->broker.id, gi->broker.host, gi->broker.port);
+ if (gi->err)
+ printf(" Error: %s\n", rd_kafka_err2str(gi->err));
+ printf(" Protocol type \"%s\", protocol \"%s\", "
+ "with %d member(s):\n",
+ gi->protocol_type, gi->protocol, gi->member_cnt);
+
+ for (j = 0 ; j < gi->member_cnt ; j++) {
+ const struct rd_kafka_group_member_info *mi;
+ mi = &gi->members[j];
+
+ printf(" \"%s\", client id \"%s\" on host %s\n",
+ mi->member_id, mi->client_id, mi->client_host);
+ printf(" metadata: %d bytes\n",
+ mi->member_metadata_size);
+ printf(" assignment: %d bytes\n",
+ mi->member_assignment_size);
+ }
+ printf("\n");
+ }
+
+ if (group && !grplist->group_cnt)
+ fprintf(stderr, "%% No matching group (%s)\n", group);
+
+ rd_kafka_group_list_destroy(grplist);
+
+ return 0;
+}
+
+
+
+static void sig_usr1 (int sig) {
+ rd_kafka_dump(stdout, rk);
+}
+
+int main (int argc, char **argv) {
+ char mode = 'C';
+ char *brokers = "localhost:9092";
+ int opt;
+ rd_kafka_conf_t *conf;
+ rd_kafka_topic_conf_t *topic_conf;
+ char errstr[512];
+ const char *debug = NULL;
+ int do_conf_dump = 0;
+ char tmp[16];
+ rd_kafka_resp_err_t err;
+ char *group = NULL;
+ rd_kafka_topic_partition_list_t *topics;
+ int is_subscription;
+ int i;
+
+ quiet = !isatty(STDIN_FILENO);
+
+ /* 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, "g:b:qd:eX:ADO")) != -1) {
+ switch (opt) {
+ case 'b':
+ brokers = optarg;
+ break;
+ case 'g':
+ group = optarg;
+ 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;
+
+ case 'D':
+ case 'O':
+ mode = opt;
+ break;
+
+ default:
+ goto usage;
+ }
+ }
+
+
+ if (do_conf_dump) {
+ const char **arr;
+ size_t cnt;
+ int pass;
+
+ for (pass = 0 ; pass < 2 ; pass++) {
+ 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 (strchr("OC", mode) && optind == argc) {
+ usage:
+ fprintf(stderr,
+ "Usage: %s [options] <topic[:part]> <topic[:part]>..\n"
+ "\n"
+ "librdkafka version %s (0x%08x)\n"
+ "\n"
+ " Options:\n"
+ " -g <group> Consumer group (%s)\n"
+ " -b <brokers> Broker address (%s)\n"
+ " -e Exit consumer when last message\n"
+ " in partition has been received.\n"
+ " -D Describe group.\n"
+ " -O Get commmitted offset(s)\n"
+ " -d [facs..] Enable debugging contexts:\n"
+ " %s\n"
+ " -q Be quiet\n"
+ " -A Raw payload output (consumer)\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"
+ "For balanced consumer groups use the 'topic1 topic2..'"
+ " format\n"
+ "and for static assignment use "
+ "'topic1:part1 topic1:part2 topic2:part1..'\n"
+ "\n",
+ argv[0],
+ rd_kafka_version_str(), rd_kafka_version(),
+ group, brokers,
+ 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);
+ }
+
+ /*
+ * Client/Consumer group
+ */
+
+ if (strchr("CO", mode)) {
+ /* Consumer groups require a group id */
+ if (!group)
+ group = "rdkafka_consumer_example";
+ if (rd_kafka_conf_set(conf, "group.id", group,
+ errstr, sizeof(errstr)) !=
+ RD_KAFKA_CONF_OK) {
+ fprintf(stderr, "%% %s\n", errstr);
+ exit(1);
+ }
+
+ /* Consumer groups always use broker based offset storage */
+ if (rd_kafka_topic_conf_set(topic_conf, "offset.store.method",
+ "broker",
+ errstr, sizeof(errstr)) !=
+ RD_KAFKA_CONF_OK) {
+ fprintf(stderr, "%% %s\n", errstr);
+ exit(1);
+ }
+
+ /* Set default topic config for pattern-matched topics. */
+ rd_kafka_conf_set_default_topic_conf(conf, topic_conf);
+
+ /* Callback called on partition assignment changes */
+ rd_kafka_conf_set_rebalance_cb(conf, rebalance_cb);
+ }
+
+ /* 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 (mode == 'D') {
+ int r;
+ /* Describe groups */
+ r = describe_groups(rk, group);
+
+ rd_kafka_destroy(rk);
+ exit(r == -1 ? 1 : 0);
+ }
+
+ /* Redirect rd_kafka_poll() to consumer_poll() */
+ rd_kafka_poll_set_consumer(rk);
+
+ topics = rd_kafka_topic_partition_list_new(argc - optind);
+ is_subscription = 1;
+ for (i = optind ; i < argc ; i++) {
+ /* Parse "topic[:part] */
+ char *topic = argv[i];
+ char *t;
+ int32_t partition = -1;
+
+ if ((t = strstr(topic, ":"))) {
+ *t = '\0';
+ partition = atoi(t+1);
+ is_subscription = 0; /* is assignment */
+ wait_eof++;
+ }
+
+ rd_kafka_topic_partition_list_add(topics, topic, partition);
+ }
+
+ if (mode == 'O') {
+ /* Offset query */
+
+ err = rd_kafka_committed(rk, topics, 5000);
+ if (err) {
+ fprintf(stderr, "%% Failed to fetch offsets: %s\n",
+ rd_kafka_err2str(err));
+ exit(1);
+ }
+
+ for (i = 0 ; i < topics->cnt ; i++) {
+ rd_kafka_topic_partition_t *p = &topics->elems[i];
+ printf("Topic \"%s\" partition %"PRId32,
+ p->topic, p->partition);
+ if (p->err)
+ printf(" error %s",
+ rd_kafka_err2str(p->err));
+ else {
+ printf(" offset %"PRId64"",
+ p->offset);
+
+ if (p->metadata_size)
+ printf(" (%d bytes of metadata)",
+ (int)p->metadata_size);
+ }
+ printf("\n");
+ }
+
+ goto done;
+ }
+
+
+ if (is_subscription) {
+ fprintf(stderr, "%% Subscribing to %d topics\n", topics->cnt);
+
+ if ((err = rd_kafka_subscribe(rk, topics))) {
+ fprintf(stderr,
+ "%% Failed to start consuming topics: %s\n",
+ rd_kafka_err2str(err));
+ exit(1);
+ }
+ } else {
+ fprintf(stderr, "%% Assigning %d partitions\n", topics->cnt);
+
+ if ((err = rd_kafka_assign(rk, topics))) {
+ fprintf(stderr,
+ "%% Failed to assign partitions: %s\n",
+ rd_kafka_err2str(err));
+ }
+ }
+
+ while (run) {
+ rd_kafka_message_t *rkmessage;
+
+ rkmessage = rd_kafka_consumer_poll(rk, 1000);
+ if (rkmessage) {
+ msg_consume(rkmessage);
+ rd_kafka_message_destroy(rkmessage);
+ }
+ }
+
+done:
+ err = rd_kafka_consumer_close(rk);
+ if (err)
+ fprintf(stderr, "%% Failed to close consumer: %s\n",
+ rd_kafka_err2str(err));
+ else
+ fprintf(stderr, "%% Consumer closed\n");
+
+ rd_kafka_topic_partition_list_destroy(topics);
+
+ /* Destroy handle */
+ rd_kafka_destroy(rk);
+
+ /* 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");
+ if (run <= 0)
+ rd_kafka_dump(stdout, rk);
+
+ return 0;
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.4/examples/rdkafka_consumer_example.cpp
----------------------------------------------------------------------
diff --git a/thirdparty/librdkafka-0.11.4/examples/rdkafka_consumer_example.cpp b/thirdparty/librdkafka-0.11.4/examples/rdkafka_consumer_example.cpp
new file mode 100644
index 0000000..83da691
--- /dev/null
+++ b/thirdparty/librdkafka-0.11.4/examples/rdkafka_consumer_example.cpp
@@ -0,0 +1,485 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2014, 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 <iostream>
+#include <string>
+#include <cstdlib>
+#include <cstdio>
+#include <csignal>
+#include <cstring>
+
+#ifndef _MSC_VER
+#include <sys/time.h>
+#endif
+
+#ifdef _MSC_VER
+#include "../win32/wingetopt.h"
+#include <atltime.h>
+#elif _AIX
+#include <unistd.h>
+#else
+#include <getopt.h>
+#include <unistd.h>
+#endif
+
+/*
+ * Typically include path in a real application would be
+ * #include <librdkafka/rdkafkacpp.h>
+ */
+#include "rdkafkacpp.h"
+
+
+
+static bool run = true;
+static bool exit_eof = false;
+static int eof_cnt = 0;
+static int partition_cnt = 0;
+static int verbosity = 1;
+static long msg_cnt = 0;
+static int64_t msg_bytes = 0;
+static void sigterm (int sig) {
+ run = false;
+}
+
+
+/**
+ * @brief format a string timestamp from the current time
+ */
+static void print_time () {
+#ifndef _MSC_VER
+ struct timeval tv;
+ char buf[64];
+ gettimeofday(&tv, NULL);
+ strftime(buf, sizeof(buf) - 1, "%Y-%m-%d %H:%M:%S", localtime(&tv.tv_sec));
+ fprintf(stderr, "%s.%03d: ", buf, (int)(tv.tv_usec / 1000));
+#else
+ std::wcerr << CTime::GetCurrentTime().Format(_T("%Y-%m-%d %H:%M:%S")).GetString()
+ << ": ";
+#endif
+}
+class ExampleEventCb : public RdKafka::EventCb {
+ public:
+ void event_cb (RdKafka::Event &event) {
+
+ print_time();
+
+ switch (event.type())
+ {
+ case RdKafka::Event::EVENT_ERROR:
+ std::cerr << "ERROR (" << RdKafka::err2str(event.err()) << "): " <<
+ event.str() << std::endl;
+ if (event.err() == RdKafka::ERR__ALL_BROKERS_DOWN)
+ run = false;
+ break;
+
+ case RdKafka::Event::EVENT_STATS:
+ std::cerr << "\"STATS\": " << event.str() << std::endl;
+ break;
+
+ case RdKafka::Event::EVENT_LOG:
+ fprintf(stderr, "LOG-%i-%s: %s\n",
+ event.severity(), event.fac().c_str(), event.str().c_str());
+ break;
+
+ case RdKafka::Event::EVENT_THROTTLE:
+ std::cerr << "THROTTLED: " << event.throttle_time() << "ms by " <<
+ event.broker_name() << " id " << (int)event.broker_id() << std::endl;
+ break;
+
+ default:
+ std::cerr << "EVENT " << event.type() <<
+ " (" << RdKafka::err2str(event.err()) << "): " <<
+ event.str() << std::endl;
+ break;
+ }
+ }
+};
+
+
+class ExampleRebalanceCb : public RdKafka::RebalanceCb {
+private:
+ static void part_list_print (const std::vector<RdKafka::TopicPartition*>&partitions){
+ for (unsigned int i = 0 ; i < partitions.size() ; i++)
+ std::cerr << partitions[i]->topic() <<
+ "[" << partitions[i]->partition() << "], ";
+ std::cerr << "\n";
+ }
+
+public:
+ void rebalance_cb (RdKafka::KafkaConsumer *consumer,
+ RdKafka::ErrorCode err,
+ std::vector<RdKafka::TopicPartition*> &partitions) {
+ std::cerr << "RebalanceCb: " << RdKafka::err2str(err) << ": ";
+
+ part_list_print(partitions);
+
+ if (err == RdKafka::ERR__ASSIGN_PARTITIONS) {
+ consumer->assign(partitions);
+ partition_cnt = (int)partitions.size();
+ } else {
+ consumer->unassign();
+ partition_cnt = 0;
+ }
+ eof_cnt = 0;
+ }
+};
+
+
+void msg_consume(RdKafka::Message* message, void* opaque) {
+ switch (message->err()) {
+ case RdKafka::ERR__TIMED_OUT:
+ break;
+
+ case RdKafka::ERR_NO_ERROR:
+ /* Real message */
+ msg_cnt++;
+ msg_bytes += message->len();
+ if (verbosity >= 3)
+ std::cerr << "Read msg at offset " << message->offset() << std::endl;
+ RdKafka::MessageTimestamp ts;
+ ts = message->timestamp();
+ if (verbosity >= 2 &&
+ ts.type != RdKafka::MessageTimestamp::MSG_TIMESTAMP_NOT_AVAILABLE) {
+ std::string tsname = "?";
+ if (ts.type == RdKafka::MessageTimestamp::MSG_TIMESTAMP_CREATE_TIME)
+ tsname = "create time";
+ else if (ts.type == RdKafka::MessageTimestamp::MSG_TIMESTAMP_LOG_APPEND_TIME)
+ tsname = "log append time";
+ std::cout << "Timestamp: " << tsname << " " << ts.timestamp << std::endl;
+ }
+ if (verbosity >= 2 && message->key()) {
+ std::cout << "Key: " << *message->key() << std::endl;
+ }
+ if (verbosity >= 1) {
+ printf("%.*s\n",
+ static_cast<int>(message->len()),
+ static_cast<const char *>(message->payload()));
+ }
+ break;
+
+ case RdKafka::ERR__PARTITION_EOF:
+ /* Last message */
+ if (exit_eof && ++eof_cnt == partition_cnt) {
+ std::cerr << "%% EOF reached for all " << partition_cnt <<
+ " partition(s)" << std::endl;
+ run = false;
+ }
+ break;
+
+ case RdKafka::ERR__UNKNOWN_TOPIC:
+ case RdKafka::ERR__UNKNOWN_PARTITION:
+ std::cerr << "Consume failed: " << message->errstr() << std::endl;
+ run = false;
+ break;
+
+ default:
+ /* Errors */
+ std::cerr << "Consume failed: " << message->errstr() << std::endl;
+ run = false;
+ }
+}
+
+
+class ExampleConsumeCb : public RdKafka::ConsumeCb {
+ public:
+ void consume_cb (RdKafka::Message &msg, void *opaque) {
+ msg_consume(&msg, opaque);
+ }
+};
+
+
+
+int main (int argc, char **argv) {
+ std::string brokers = "localhost";
+ std::string errstr;
+ std::string topic_str;
+ std::string mode;
+ std::string debug;
+ std::vector<std::string> topics;
+ bool do_conf_dump = false;
+ int opt;
+ int use_ccb = 0;
+
+ /*
+ * Create configuration objects
+ */
+ RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
+ RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
+
+ ExampleRebalanceCb ex_rebalance_cb;
+ conf->set("rebalance_cb", &ex_rebalance_cb, errstr);
+
+ while ((opt = getopt(argc, argv, "g:b:z:qd:eX:AM:f:qv")) != -1) {
+ switch (opt) {
+ case 'g':
+ if (conf->set("group.id", optarg, errstr) != RdKafka::Conf::CONF_OK) {
+ std::cerr << errstr << std::endl;
+ exit(1);
+ }
+ break;
+ case 'b':
+ brokers = optarg;
+ break;
+ case 'z':
+ if (conf->set("compression.codec", optarg, errstr) !=
+ RdKafka::Conf::CONF_OK) {
+ std::cerr << errstr << std::endl;
+ exit(1);
+ }
+ break;
+ case 'e':
+ exit_eof = true;
+ break;
+ case 'd':
+ debug = optarg;
+ break;
+ case 'M':
+ if (conf->set("statistics.interval.ms", optarg, errstr) !=
+ RdKafka::Conf::CONF_OK) {
+ std::cerr << errstr << std::endl;
+ exit(1);
+ }
+ break;
+ case 'X':
+ {
+ char *name, *val;
+
+ if (!strcmp(optarg, "dump")) {
+ do_conf_dump = true;
+ continue;
+ }
+
+ name = optarg;
+ if (!(val = strchr(name, '='))) {
+ std::cerr << "%% Expected -X property=value, not " <<
+ name << std::endl;
+ exit(1);
+ }
+
+ *val = '\0';
+ val++;
+
+ /* Try "topic." prefixed properties on topic
+ * conf first, and then fall through to global if
+ * it didnt match a topic configuration property. */
+ RdKafka::Conf::ConfResult res = RdKafka::Conf::CONF_UNKNOWN;
+ if (!strncmp(name, "topic.", strlen("topic.")))
+ res = tconf->set(name+strlen("topic."), val, errstr);
+ if (res == RdKafka::Conf::CONF_UNKNOWN)
+ res = conf->set(name, val, errstr);
+
+ if (res != RdKafka::Conf::CONF_OK) {
+ std::cerr << errstr << std::endl;
+ exit(1);
+ }
+ }
+ break;
+
+ case 'f':
+ if (!strcmp(optarg, "ccb"))
+ use_ccb = 1;
+ else {
+ std::cerr << "Unknown option: " << optarg << std::endl;
+ exit(1);
+ }
+ break;
+
+ case 'q':
+ verbosity--;
+ break;
+
+ case 'v':
+ verbosity++;
+ break;
+
+ default:
+ goto usage;
+ }
+ }
+
+ for (; optind < argc ; optind++)
+ topics.push_back(std::string(argv[optind]));
+
+ if (topics.empty() || optind != argc) {
+ usage:
+ fprintf(stderr,
+ "Usage: %s -g <group-id> [options] topic1 topic2..\n"
+ "\n"
+ "librdkafka version %s (0x%08x)\n"
+ "\n"
+ " Options:\n"
+ " -g <group-id> Consumer group id\n"
+ " -b <brokers> Broker address (localhost:9092)\n"
+ " -z <codec> Enable compression:\n"
+ " none|gzip|snappy\n"
+ " -e Exit consumer when last message\n"
+ " in partition has been received.\n"
+ " -d [facs..] Enable debugging contexts:\n"
+ " %s\n"
+ " -M <intervalms> Enable statistics\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"
+ " -f <flag> Set option:\n"
+ " ccb - use consume_callback\n"
+ " -q Quiet / Decrease verbosity\n"
+ " -v Increase verbosity\n"
+ "\n"
+ "\n",
+ argv[0],
+ RdKafka::version_str().c_str(), RdKafka::version(),
+ RdKafka::get_debug_contexts().c_str());
+ exit(1);
+ }
+
+
+ /*
+ * Set configuration properties
+ */
+ conf->set("metadata.broker.list", brokers, errstr);
+
+ if (!debug.empty()) {
+ if (conf->set("debug", debug, errstr) != RdKafka::Conf::CONF_OK) {
+ std::cerr << errstr << std::endl;
+ exit(1);
+ }
+ }
+
+ ExampleConsumeCb ex_consume_cb;
+
+ if(use_ccb) {
+ conf->set("consume_cb", &ex_consume_cb, errstr);
+ }
+
+ ExampleEventCb ex_event_cb;
+ conf->set("event_cb", &ex_event_cb, errstr);
+
+ if (do_conf_dump) {
+ int pass;
+
+ for (pass = 0 ; pass < 2 ; pass++) {
+ std::list<std::string> *dump;
+ if (pass == 0) {
+ dump = conf->dump();
+ std::cout << "# Global config" << std::endl;
+ } else {
+ dump = tconf->dump();
+ std::cout << "# Topic config" << std::endl;
+ }
+
+ for (std::list<std::string>::iterator it = dump->begin();
+ it != dump->end(); ) {
+ std::cout << *it << " = ";
+ it++;
+ std::cout << *it << std::endl;
+ it++;
+ }
+ std::cout << std::endl;
+ }
+ exit(0);
+ }
+
+ conf->set("default_topic_conf", tconf, errstr);
+ delete tconf;
+
+ signal(SIGINT, sigterm);
+ signal(SIGTERM, sigterm);
+
+
+ /*
+ * Consumer mode
+ */
+
+ /*
+ * Create consumer using accumulated global configuration.
+ */
+ RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);
+ if (!consumer) {
+ std::cerr << "Failed to create consumer: " << errstr << std::endl;
+ exit(1);
+ }
+
+ delete conf;
+
+ std::cout << "% Created consumer " << consumer->name() << std::endl;
+
+
+ /*
+ * Subscribe to topics
+ */
+ RdKafka::ErrorCode err = consumer->subscribe(topics);
+ if (err) {
+ std::cerr << "Failed to subscribe to " << topics.size() << " topics: "
+ << RdKafka::err2str(err) << std::endl;
+ exit(1);
+ }
+
+ /*
+ * Consume messages
+ */
+ while (run) {
+ RdKafka::Message *msg = consumer->consume(1000);
+ if (!use_ccb) {
+ msg_consume(msg, NULL);
+ }
+ delete msg;
+ }
+
+#ifndef _MSC_VER
+ alarm(10);
+#endif
+
+ /*
+ * Stop consumer
+ */
+ consumer->close();
+ delete consumer;
+
+ std::cerr << "% Consumed " << msg_cnt << " messages ("
+ << msg_bytes << " bytes)" << std::endl;
+
+ /*
+ * Wait for RdKafka to decommission.
+ * This is not strictly needed (with check outq_len() above), but
+ * allows RdKafka to clean up all its resources before the application
+ * exits so that memory profilers such as valgrind wont complain about
+ * memory leaks.
+ */
+ RdKafka::wait_destroyed(5000);
+
+ return 0;
+}