1 files changed, 197 insertions, 0 deletions
diff --git a/fluent-bit/lib/librdkafka-2.1.0/tests/0070-null_empty.cpp b/fluent-bit/lib/librdkafka-2.1.0/tests/0070-null_empty.cpp
new file mode 100644
index 000000000..fac48185c
--- /dev/null
+++ b/fluent-bit/lib/librdkafka-2.1.0/tests/0070-null_empty.cpp
@@ -0,0 +1,197 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2016, Magnus Edenhill
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+
+#include "testcpp.h"
+#include <cstring>
+
+/**
+ * Verification of difference between empty and null Key and Value
+ */
+
+
+static int check_equal(const char *exp,
+                       const char *actual,
+                       size_t len,
+                       std::string what) {
+  size_t exp_len = exp ? strlen(exp) : 0;
+  int failures   = 0;
+
+  if (!actual && len != 0) {
+    Test::FailLater(tostr()
+                    << what << ": expected length 0 for Null, not " << len);
+    failures++;
+  }
+
+  if (exp) {
+    if (!actual) {
+      Test::FailLater(tostr()
+                      << what << ": expected \"" << exp << "\", not Null");
+      failures++;
+
+    } else if (len != exp_len || strncmp(exp, actual, exp_len)) {
+      Test::FailLater(tostr() << what << ": expected \"" << exp << "\", not \""
+                              << actual << "\" (" << len << " bytes)");
+      failures++;
+    }
+
+  } else {
+    if (actual) {
+      Test::FailLater(tostr() << what << ": expected Null, not \"" << actual
+                              << "\" (" << len << " bytes)");
+      failures++;
+    }
+  }
+
+  if (!failures)
+    Test::Say(3, tostr() << what << ": matched expectation\n");
+
+  return failures;
+}
+
+
+static void do_test_null_empty(bool api_version_request) {
+  std::string topic   = Test::mk_topic_name("0070_null_empty", 1);
+  const int partition = 0;
+
+  Test::Say(tostr() << "Testing with api.version.request="
+                    << api_version_request << " on topic " << topic
+                    << " partition " << partition << "\n");
+
+  RdKafka::Conf *conf;
+  Test::conf_init(&conf, NULL, 0);
+  Test::conf_set(conf, "api.version.request",
+                 api_version_request ? "true" : "false");
+  Test::conf_set(conf, "acks", "all");
+
+
+  std::string errstr;
+  RdKafka::Producer *p = RdKafka::Producer::create(conf, errstr);
+  if (!p)
+    Test::Fail("Failed to create Producer: " + errstr);
+  delete conf;
+
+  const int msgcnt                    = 8;
+  static const char *msgs[msgcnt * 2] = {NULL,   NULL, "key2", NULL, "key3",
+                                         "val3", NULL, "val4", "",   NULL,
+                                         NULL,   "",   "",     ""};
+
+  RdKafka::ErrorCode err;
+
+  for (int i = 0; i < msgcnt * 2; i += 2) {
+    Test::Say(3, tostr() << "Produce message #" << (i / 2) << ": key=\""
+                         << (msgs[i] ? msgs[i] : "Null") << "\", value=\""
+                         << (msgs[i + 1] ? msgs[i + 1] : "Null") << "\"\n");
+    err = p->produce(topic, partition, RdKafka::Producer::RK_MSG_COPY,
+                     /* Value */
+                     (void *)msgs[i + 1], msgs[i + 1] ? strlen(msgs[i + 1]) : 0,
+                     /* Key */
+                     (void *)msgs[i], msgs[i] ? strlen(msgs[i]) : 0, 0, NULL);
+    if (err != RdKafka::ERR_NO_ERROR)
+      Test::Fail("Produce failed: " + RdKafka::err2str(err));
+  }
+
+  if (p->flush(tmout_multip(3 * 5000)) != 0)
+    Test::Fail("Not all messages flushed");
+
+  Test::Say(tostr() << "Produced " << msgcnt << " messages to " << topic
+                    << "\n");
+
+  delete p;
+
+  /*
+   * Now consume messages from the beginning, making sure they match
+   * what was produced.
+   */
+
+  /* Create consumer */
+  Test::conf_init(&conf, NULL, 10);
+  Test::conf_set(conf, "group.id", topic);
+  Test::conf_set(conf, "api.version.request",
+                 api_version_request ? "true" : "false");
+  Test::conf_set(conf, "enable.auto.commit", "false");
+
+  RdKafka::KafkaConsumer *c = RdKafka::KafkaConsumer::create(conf, errstr);
+  if (!c)
+    Test::Fail("Failed to create KafkaConsumer: " + errstr);
+  delete conf;
+
+  /* Assign the partition */
+  std::vector<RdKafka::TopicPartition *> parts;
+  parts.push_back(RdKafka::TopicPartition::create(
+      topic, partition, RdKafka::Topic::OFFSET_BEGINNING));
+  err = c->assign(parts);
+  if (err != RdKafka::ERR_NO_ERROR)
+    Test::Fail("assign() failed: " + RdKafka::err2str(err));
+  RdKafka::TopicPartition::destroy(parts);
+
+  /* Start consuming */
+  int failures = 0;
+  for (int i = 0; i < msgcnt * 2; i += 2) {
+    RdKafka::Message *msg = c->consume(tmout_multip(5000));
+    if (msg->err())
+      Test::Fail(tostr() << "consume() failed at message " << (i / 2) << ": "
+                         << msg->errstr());
+
+    /* verify key */
+    failures += check_equal(msgs[i], msg->key() ? msg->key()->c_str() : NULL,
+                            msg->key_len(),
+                            tostr() << "message #" << (i / 2) << " (offset "
+                                    << msg->offset() << ") key");
+    /* verify key_pointer() API as too */
+    failures +=
+        check_equal(msgs[i], (const char *)msg->key_pointer(), msg->key_len(),
+                    tostr() << "message #" << (i / 2) << " (offset "
+                            << msg->offset() << ") key");
+
+    /* verify value */
+    failures +=
+        check_equal(msgs[i + 1], (const char *)msg->payload(), msg->len(),
+                    tostr() << "message #" << (i / 2) << " (offset "
+                            << msg->offset() << ") value");
+    delete msg;
+  }
+
+  Test::Say(tostr() << "Done consuming, closing. " << failures
+                    << " test failures\n");
+  if (failures)
+    Test::Fail(tostr() << "See " << failures << "  previous test failure(s)");
+
+  c->close();
+  delete c;
+}
+
+
+extern "C" {
+int main_0070_null_empty(int argc, char **argv) {
+  if (test_broker_version >= TEST_BRKVER(0, 10, 0, 0))
+    do_test_null_empty(true);
+  do_test_null_empty(false);
+  return 0;
+}
+}