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

#if WITH_RAPIDJSON

#include <iostream>
#include <cstring>
#include <cstdlib>
#include <assert.h>
#include <sstream>
#include <string>
#include <map>

#include <rapidjson/document.h>
#include <rapidjson/schema.h>
#include <rapidjson/filereadstream.h>
#include <rapidjson/stringbuffer.h>
#include <rapidjson/error/en.h>
#include <rapidjson/prettywriter.h>


/**
 * @name Consumer Transactions.
 *
 * - Uses the TransactionProducerCli Java application to produce messages
 *   that are part of abort and commit transactions in various combinations
 *   and tests that librdkafka consumes them as expected. Refer to
 *   TransactionProducerCli.java for scenarios covered.
 */


class TestEventCb : public RdKafka::EventCb {
 public:
  static bool should_capture_stats;
  static bool has_captured_stats;
  static int64_t partition_0_hi_offset;
  static int64_t partition_0_ls_offset;
  static std::string topic;

  void event_cb(RdKafka::Event &event) {
    switch (event.type()) {
    case RdKafka::Event::EVENT_STATS:
      if (should_capture_stats) {
        partition_0_hi_offset = -1;
        partition_0_ls_offset = -1;

        has_captured_stats   = true;
        should_capture_stats = false;
        char path[256];

        /* Parse JSON to validate */
        rapidjson::Document d;
        if (d.Parse(event.str().c_str()).HasParseError())
          Test::Fail(tostr() << "Failed to parse stats JSON: "
                             << rapidjson::GetParseError_En(d.GetParseError())
                             << " at " << d.GetErrorOffset());

        rd_snprintf(path, sizeof(path), "/topics/%s/partitions/0",
                    topic.c_str());

        rapidjson::Pointer jpath((const char *)path);
        rapidjson::Value *pp = rapidjson::GetValueByPointer(d, jpath);
        if (pp == NULL)
          return;

        TEST_ASSERT(pp->HasMember("hi_offset"), "hi_offset not found in stats");
        TEST_ASSERT(pp->HasMember("ls_offset"), "ls_offset not found in stats");

        partition_0_hi_offset = (*pp)["hi_offset"].GetInt();
        partition_0_ls_offset = (*pp)["ls_offset"].GetInt();
      }
      break;

    case RdKafka::Event::EVENT_LOG:
      std::cerr << event.str() << "\n";
      break;

    default:
      break;
    }
  }
};

bool TestEventCb::should_capture_stats;
bool TestEventCb::has_captured_stats;
int64_t TestEventCb::partition_0_hi_offset;
int64_t TestEventCb::partition_0_ls_offset;
std::string TestEventCb::topic;

static TestEventCb ex_event_cb;


static void execute_java_produce_cli(std::string &bootstrapServers,
                                     const std::string &topic,
                                     const std::string &testidstr,
                                     const char **cmds,
                                     size_t cmd_cnt) {
  const std::string topicCmd  = "topic," + topic;
  const std::string testidCmd = "testid," + testidstr;
  const char **argv;
  size_t i = 0;

  argv = (const char **)rd_alloca(sizeof(*argv) * (1 + 1 + 1 + cmd_cnt + 1));
  argv[i++] = bootstrapServers.c_str();
  argv[i++] = topicCmd.c_str();
  argv[i++] = testidCmd.c_str();

  for (size_t j = 0; j < cmd_cnt; j++)
    argv[i++] = cmds[j];

  argv[i] = NULL;

  int pid = test_run_java("TransactionProducerCli", (const char **)argv);
  test_waitpid(pid);
}

static std::vector<RdKafka::Message *>
consume_messages(RdKafka::KafkaConsumer *c, std::string topic, int partition) {
  RdKafka::ErrorCode err;

  /* Assign partitions */
  std::vector<RdKafka::TopicPartition *> parts;
  parts.push_back(RdKafka::TopicPartition::create(topic, partition));
  if ((err = c->assign(parts)))
    Test::Fail("assign failed: " + RdKafka::err2str(err));
  RdKafka::TopicPartition::destroy(parts);

  Test::Say(tostr() << "Consuming from topic " << topic << " partition "
                    << partition << "\n");
  std::vector<RdKafka::Message *> result = std::vector<RdKafka::Message *>();

  while (true) {
    RdKafka::Message *msg = c->consume(tmout_multip(1000));
    switch (msg->err()) {
    case RdKafka::ERR__TIMED_OUT:
      delete msg;
      continue;
    case RdKafka::ERR__PARTITION_EOF:
      delete msg;
      break;
    case RdKafka::ERR_NO_ERROR:
      result.push_back(msg);
      continue;
    default:
      Test::Fail("Error consuming from topic " + topic + ": " + msg->errstr());
      delete msg;
      break;
    }
    break;
  }

  Test::Say("Read all messages from topic: " + topic + "\n");

  TestEventCb::should_capture_stats = true;

  /* rely on the test timeout to prevent an infinite loop in
   * the (unlikely) event that the statistics callback isn't
   * called. */
  while (!TestEventCb::has_captured_stats) {
    RdKafka::Message *msg = c->consume(tmout_multip(500));
    delete msg;
  }

  Test::Say("Captured consumer statistics event\n");

  return result;
}


static void delete_messages(std::vector<RdKafka::Message *> &messages) {
  for (size_t i = 0; i < messages.size(); ++i)
    delete messages[i];
}


static std::string get_bootstrap_servers() {
  RdKafka::Conf *conf;
  std::string bootstrap_servers;
  Test::conf_init(&conf, NULL, 40);
  conf->get("bootstrap.servers", bootstrap_servers);
  delete conf;
  return bootstrap_servers;
}


static RdKafka::KafkaConsumer *create_consumer(std::string &topic_name,
                                               const char *isolation_level) {
  RdKafka::Conf *conf;
  std::string errstr;

  Test::conf_init(&conf, NULL, 40);
  Test::conf_set(conf, "group.id", topic_name);
  Test::conf_set(conf, "enable.auto.commit", "false");
  Test::conf_set(conf, "auto.offset.reset", "earliest");
  Test::conf_set(conf, "enable.partition.eof", "true");
  Test::conf_set(conf, "isolation.level", isolation_level);
  Test::conf_set(conf, "statistics.interval.ms", "1000");
  conf->set("event_cb", &ex_event_cb, errstr);
  TestEventCb::should_capture_stats = false;
  TestEventCb::has_captured_stats   = false;

  RdKafka::KafkaConsumer *c = RdKafka::KafkaConsumer::create(conf, errstr);
  if (!c)
    Test::Fail("Failed to create KafkaConsumer: " + errstr);

  delete conf;

  return c;
}


static std::vector<std::string> csv_split(const std::string &input) {
  std::stringstream ss(input);
  std::vector<std::string> res;

  while (ss.good()) {
    std::string substr;
    std::getline(ss, substr, ',');
    /* Trim */
    substr.erase(0, substr.find_first_not_of(' '));
    substr.erase(substr.find_last_not_of(' ') + 1);
    res.push_back(substr);
  }

  return res;
}



enum TransactionType {
  TransactionType_None,
  TransactionType_BeginAbort,
  TransactionType_BeginCommit,
  TransactionType_BeginOpen,
  TransactionType_ContinueAbort,
  TransactionType_ContinueCommit,
  TransactionType_ContinueOpen
};

static TransactionType TransactionType_from_string(std::string str) {
#define _CHKRET(NAME)                                                          \
  if (!str.compare(#NAME))                                                     \
  return TransactionType_##NAME

  _CHKRET(None);
  _CHKRET(BeginAbort);
  _CHKRET(BeginCommit);
  _CHKRET(BeginOpen);
  _CHKRET(ContinueAbort);
  _CHKRET(ContinueCommit);
  _CHKRET(ContinueOpen);

  Test::Fail("Unknown TransactionType: " + str);

  return TransactionType_None; /* NOTREACHED */
}


static void txn_producer_makeTestMessages(RdKafka::Producer *producer,
                                          const std::string &topic,
                                          const std::string &testidstr,
                                          int partition,
                                          int idStart,
                                          int msgcount,
                                          TransactionType tt,
                                          bool do_flush) {
  RdKafka::Error *error;

  if (tt != TransactionType_None && tt != TransactionType_ContinueOpen &&
      tt != TransactionType_ContinueCommit &&
      tt != TransactionType_ContinueAbort) {
    error = producer->begin_transaction();
    if (error) {
      Test::Fail("begin_transaction() failed: " + error->str());
      delete error;
    }
  }

  for (int i = 0; i < msgcount; i++) {
    char key[]     = {(char)((i + idStart) & 0xff)};
    char payload[] = {0x10, 0x20, 0x30, 0x40};
    RdKafka::ErrorCode err;

    err = producer->produce(topic, partition, producer->RK_MSG_COPY, payload,
                            sizeof(payload), key, sizeof(key), 0, NULL);
    if (err)
      Test::Fail("produce() failed: " + RdKafka::err2str(err));
  }

  if (do_flush)
    producer->flush(-1);

  switch (tt) {
  case TransactionType_BeginAbort:
  case TransactionType_ContinueAbort:
    error = producer->abort_transaction(30 * 1000);
    if (error) {
      Test::Fail("abort_transaction() failed: " + error->str());
      delete error;
    }
    break;

  case TransactionType_BeginCommit:
  case TransactionType_ContinueCommit:
    error = producer->commit_transaction(30 * 1000);
    if (error) {
      Test::Fail("commit_transaction() failed: " + error->str());
      delete error;
    }
    break;

  default:
    break;
  }
}


class txnDeliveryReportCb : public RdKafka::DeliveryReportCb {
 public:
  void dr_cb(RdKafka::Message &msg) {
    switch (msg.err()) {
    case RdKafka::ERR__PURGE_QUEUE:
    case RdKafka::ERR__PURGE_INFLIGHT:
      /* These are expected when transactions are aborted */
      break;

    case RdKafka::ERR_NO_ERROR:
      break;

    default:
      Test::Fail("Delivery failed: " + msg.errstr());
      break;
    }
  }
};


/**
 * @brief Transactional producer, performing the commands in \p cmds.
 *        This is the librdkafka counterpart of
 *        java/TransactionProducerCli.java
 */
static void txn_producer(const std::string &brokers,
                         const std::string &topic,
                         const std::string &testidstr,
                         const char **cmds,
                         size_t cmd_cnt) {
  RdKafka::Conf *conf;
  txnDeliveryReportCb txn_dr;

  Test::conf_init(&conf, NULL, 0);
  Test::conf_set(conf, "bootstrap.servers", brokers);


  std::map<std::string, RdKafka::Producer *> producers;

  for (size_t i = 0; i < cmd_cnt; i++) {
    std::string cmdstr = std::string(cmds[i]);

    Test::Say(_C_CLR "rdkafka txn producer command: " + cmdstr + "\n");

    std::vector<std::string> cmd = csv_split(cmdstr);

    if (!cmd[0].compare("sleep")) {
      rd_usleep(atoi(cmd[1].c_str()) * 1000, NULL);

    } else if (!cmd[0].compare("exit")) {
      break; /* We can't really simulate the Java exit behaviour
              * from in-process. */

    } else if (cmd[0].find("producer") == 0) {
      TransactionType txntype = TransactionType_from_string(cmd[4]);

      std::map<std::string, RdKafka::Producer *>::iterator it =
          producers.find(cmd[0]);

      RdKafka::Producer *producer;

      if (it == producers.end()) {
        /* Create producer if it doesn't exist */
        std::string errstr;

        Test::Say(tostr() << "Creating producer " << cmd[0]
                          << " with transactiontype " << txntype << " '"
                          << cmd[4] << "'\n");

        /* Config */
        Test::conf_set(conf, "enable.idempotence", "true");
        if (txntype != TransactionType_None)
          Test::conf_set(conf, "transactional.id",
                         "test-transactional-id-c-" + testidstr + "-" + cmd[0]);
        else
          Test::conf_set(conf, "transactional.id", "");
        Test::conf_set(conf, "linger.ms", "5"); /* ensure batching */
        conf->set("dr_cb", &txn_dr, errstr);

        /* Create producer */
        producer = RdKafka::Producer::create(conf, errstr);
        if (!producer)
          Test::Fail("Failed to create producer " + cmd[0] + ": " + errstr);

        /* Init transactions if producer is transactional */
        if (txntype != TransactionType_None) {
          RdKafka::Error *error = producer->init_transactions(20 * 1000);
          if (error) {
            Test::Fail("init_transactions() failed: " + error->str());
            delete error;
          }
        }


        producers[cmd[0]] = producer;
      } else {
        producer = it->second;
      }

      txn_producer_makeTestMessages(
          producer,                             /* producer */
          topic,                                /* topic */
          testidstr,                            /* testid */
          atoi(cmd[1].c_str()),                 /* partition */
          (int)strtol(cmd[2].c_str(), NULL, 0), /* idStart */
          atoi(cmd[3].c_str()),                 /* msg count */
          txntype,                              /* TransactionType */
          !cmd[5].compare("DoFlush") /* Flush */);

    } else {
      Test::Fail("Unknown command: " + cmd[0]);
    }
  }

  delete conf;

  for (std::map<std::string, RdKafka::Producer *>::iterator it =
           producers.begin();
       it != producers.end(); it++)
    delete it->second;
}



static void do_test_consumer_txn_test(bool use_java_producer) {
  std::string errstr;
  std::string topic_name;
  RdKafka::KafkaConsumer *c;
  std::vector<RdKafka::Message *> msgs;
  std::string testidstr = test_str_id_generate_tmp();

  std::string bootstrap_servers = get_bootstrap_servers();

  Test::Say(tostr() << _C_BLU "[ Consumer transaction tests using "
                    << (use_java_producer ? "java" : "librdkafka")
                    << " producer with testid " << testidstr << "]\n" _C_CLR);

#define run_producer(CMDS...)                                                  \
  do {                                                                         \
    const char *_cmds[] = {CMDS};                                              \
    size_t _cmd_cnt     = sizeof(_cmds) / sizeof(*_cmds);                      \
    if (use_java_producer)                                                     \
      execute_java_produce_cli(bootstrap_servers, topic_name, testidstr,       \
                               _cmds, _cmd_cnt);                               \
    else                                                                       \
      txn_producer(bootstrap_servers, topic_name, testidstr, _cmds, _cmd_cnt); \
  } while (0)

  if (test_quick) {
    Test::Say("Skipping consumer_txn tests 0->4 due to quick mode\n");
    goto test5;
  }


  Test::Say(_C_BLU "Test 0 - basic commit + abort\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-0", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer1, -1, 0x0, 5, BeginCommit, DoFlush",
               "producer1, -1, 0x10, 5, BeginAbort, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 5,
              "Consumed unexpected number of messages. "
              "Expected 5, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 && 0 == msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 && 4 == msgs[4]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);
  c->close();
  delete c;

#define expect_msgcnt(msgcnt)                                                  \
  TEST_ASSERT(msgs.size() == msgcnt, "Expected %d messages, got %d",           \
              (int)msgs.size(), msgcnt)

#define expect_key(msgidx, value)                                              \
  do {                                                                         \
    TEST_ASSERT(msgs.size() > msgidx,                                          \
                "Expected at least %d message(s), only got %d", msgidx + 1,    \
                (int)msgs.size());                                             \
    TEST_ASSERT(msgs[msgidx]->key_len() == 1,                                  \
                "Expected msg #%d key to be of size 1, not %d\n", msgidx,      \
                (int)msgs[msgidx]->key_len());                                 \
    TEST_ASSERT(value == (int)msgs[msgidx]->key()->c_str()[0],                 \
                "Expected msg #%d key 0x%x, not 0x%x", msgidx, value,          \
                (int)msgs[msgidx]->key()->c_str()[0]);                         \
  } while (0)

  c    = create_consumer(topic_name, "READ_UNCOMMITTED");
  msgs = consume_messages(c, topic_name, 0);
  expect_msgcnt(10);
  expect_key(0, 0x0);
  expect_key(4, 0x4);
  expect_key(5, 0x10);
  expect_key(9, 0x14);
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 0.1\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-0.1", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer1, -1, 0x0, 5, BeginCommit, DontFlush",
               "producer1, -1, 0x10, 5, BeginAbort, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 5,
              "Consumed unexpected number of messages. "
              "Expected 5, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 && 0 == msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 && 4 == msgs[4]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);
  c->close();
  delete c;

  c    = create_consumer(topic_name, "READ_UNCOMMITTED");
  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 10,
              "Consumed unexpected number of messages. "
              "Expected 10, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 && 0 == msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 && 4 == msgs[4]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[5]->key_len() >= 1 && 0x10 == msgs[5]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[9]->key_len() >= 1 && 0x14 == msgs[9]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 0.2\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-0.2", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer1, -1, 0x10, 5, BeginAbort, DoFlush",
               "producer1, -1, 0x30, 5, BeginCommit, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 5,
              "Consumed unexpected number of messages. "
              "Expected 5, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 && 0x30 == msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 && 0x34 == msgs[4]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);
  c->close();
  delete c;

  c    = create_consumer(topic_name, "READ_UNCOMMITTED");
  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 10,
              "Consumed unexpected number of messages. "
              "Expected 10, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 && 0x10 == msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 && 0x14 == msgs[4]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[5]->key_len() >= 1 && 0x30 == msgs[5]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[9]->key_len() >= 1 && 0x34 == msgs[9]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 1 - mixed with non-transactional.\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-1", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);
  TestEventCb::topic = topic_name;

  run_producer("producer3, -1, 0x10, 5, None, DoFlush",
               "producer1, -1, 0x50, 5, BeginCommit, DoFlush",
               "producer1, -1, 0x80, 5, BeginAbort, DoFlush");

  msgs = consume_messages(c, topic_name, 0);

  TEST_ASSERT(TestEventCb::partition_0_ls_offset != -1 &&
                  TestEventCb::partition_0_ls_offset ==
                      TestEventCb::partition_0_hi_offset,
              "Expected hi_offset to equal ls_offset but "
              "got hi_offset: %" PRId64 ", ls_offset: %" PRId64,
              TestEventCb::partition_0_hi_offset,
              TestEventCb::partition_0_ls_offset);

  TEST_ASSERT(msgs.size() == 10,
              "Consumed unexpected number of messages. "
              "Expected 10, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 && 0x10 == msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 && 0x14 == msgs[4]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[5]->key_len() >= 1 && 0x50 == msgs[5]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[9]->key_len() >= 1 && 0x54 == msgs[9]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;

  Test::Say(_C_BLU "Test 1.1\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-1.1", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer1, -1, 0x30, 5, BeginAbort, DoFlush",
               "producer3, -1, 0x40, 5, None, DoFlush",
               "producer1, -1, 0x60, 5, BeginCommit, DoFlush");


  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 10,
              "Consumed unexpected number of messages. "
              "Expected 10, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 && 0x40 == msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 && 0x44 == msgs[4]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[5]->key_len() >= 1 && 0x60 == msgs[5]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[9]->key_len() >= 1 && 0x64 == msgs[9]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 1.2\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-1.2", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer1, -1, 0x10, 5, BeginCommit, DoFlush",
               "producer1, -1, 0x20, 5, BeginAbort, DoFlush",
               "producer3, -1, 0x30, 5, None, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 10,
              "Consumed unexpected number of messages. "
              "Expected 10, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 && 0x10 == msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 && 0x14 == msgs[4]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[5]->key_len() >= 1 && 0x30 == msgs[5]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[9]->key_len() >= 1 && 0x34 == msgs[9]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 2 - rapid abort / committing.\n" _C_CLR);
  // note: aborted records never seem to make it to the broker when not flushed.

  topic_name = Test::mk_topic_name("0098-consumer_txn-2", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer1, -1, 0x10, 1, BeginAbort, DontFlush",
               "producer1, -1, 0x20, 1, BeginCommit, DontFlush",
               "producer1, -1, 0x30, 1, BeginAbort, DontFlush",
               "producer1, -1, 0x40, 1, BeginCommit, DontFlush",
               "producer1, -1, 0x50, 1, BeginAbort, DontFlush",
               "producer1, -1, 0x60, 1, BeginCommit, DontFlush",
               "producer1, -1, 0x70, 1, BeginAbort, DontFlush",
               "producer1, -1, 0x80, 1, BeginCommit, DontFlush",
               "producer1, -1, 0x90, 1, BeginAbort, DontFlush",
               "producer1, -1, 0xa0, 1, BeginCommit, DoFlush",
               "producer3, -1, 0xb0, 1, None, DontFlush",
               "producer3, -1, 0xc0, 1, None, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 7,
              "Consumed unexpected number of messages. "
              "Expected 7, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 &&
                  0x20 == (unsigned char)msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[1]->key_len() >= 1 &&
                  0x40 == (unsigned char)msgs[1]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[2]->key_len() >= 1 &&
                  0x60 == (unsigned char)msgs[2]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[3]->key_len() >= 1 &&
                  0x80 == (unsigned char)msgs[3]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 &&
                  0xa0 == (unsigned char)msgs[4]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[5]->key_len() >= 1 &&
                  0xb0 == (unsigned char)msgs[5]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[6]->key_len() >= 1 &&
                  0xc0 == (unsigned char)msgs[6]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 2.1\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-2.1", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer1, -1, 0x10, 1, BeginAbort, DoFlush",
               "producer1, -1, 0x20, 1, BeginCommit, DoFlush",
               "producer1, -1, 0x30, 1, BeginAbort, DoFlush",
               "producer1, -1, 0x40, 1, BeginCommit, DoFlush",
               "producer1, -1, 0x50, 1, BeginAbort, DoFlush",
               "producer1, -1, 0x60, 1, BeginCommit, DoFlush",
               "producer1, -1, 0x70, 1, BeginAbort, DoFlush",
               "producer1, -1, 0x80, 1, BeginCommit, DoFlush",
               "producer1, -1, 0x90, 1, BeginAbort, DoFlush",
               "producer1, -1, 0xa0, 1, BeginCommit, DoFlush",
               "producer3, -1, 0xb0, 1, None, DoFlush",
               "producer3, -1, 0xc0, 1, None, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 7,
              "Consumed unexpected number of messages. "
              "Expected 7, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 &&
                  0x20 == (unsigned char)msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[1]->key_len() >= 1 &&
                  0x40 == (unsigned char)msgs[1]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[2]->key_len() >= 1 &&
                  0x60 == (unsigned char)msgs[2]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[3]->key_len() >= 1 &&
                  0x80 == (unsigned char)msgs[3]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 &&
                  0xa0 == (unsigned char)msgs[4]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[5]->key_len() >= 1 &&
                  0xb0 == (unsigned char)msgs[5]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[6]->key_len() >= 1 &&
                  0xc0 == (unsigned char)msgs[6]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);
  c->close();
  delete c;

  c    = create_consumer(topic_name, "READ_UNCOMMITTED");
  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 12,
              "Consumed unexpected number of messages. "
              "Expected 12, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 &&
                  0x10 == (unsigned char)msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[1]->key_len() >= 1 &&
                  0x20 == (unsigned char)msgs[1]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[2]->key_len() >= 1 &&
                  0x30 == (unsigned char)msgs[2]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[3]->key_len() >= 1 &&
                  0x40 == (unsigned char)msgs[3]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 &&
                  0x50 == (unsigned char)msgs[4]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[5]->key_len() >= 1 &&
                  0x60 == (unsigned char)msgs[5]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[6]->key_len() >= 1 &&
                  0x70 == (unsigned char)msgs[6]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 3 - cross partition (simple).\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-3", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 2, 3);

  run_producer("producer1, 0, 0x10, 3, BeginOpen, DoFlush",
               "producer1, 1, 0x20, 3, ContinueOpen, DoFlush",
               "producer1, 0, 0x30, 3, ContinueCommit, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 6,
              "Consumed unexpected number of messages. "
              "Expected 6, got: %d",
              (int)msgs.size());
  delete_messages(msgs);
  msgs = consume_messages(c, topic_name, 1);
  TEST_ASSERT(msgs.size() == 3,
              "Consumed unexpected number of messages. "
              "Expected 3, got: %d",
              (int)msgs.size());
  delete_messages(msgs);
  c->close();
  delete c;

  c    = create_consumer(topic_name, "READ_UNCOMMITTED");
  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 6,
              "Consumed unexpected number of messages. "
              "Expected 6, got: %d",
              (int)msgs.size());
  delete_messages(msgs);
  msgs = consume_messages(c, topic_name, 1);
  TEST_ASSERT(msgs.size() == 3,
              "Consumed unexpected number of messages. "
              "Expected 3, got: %d",
              (int)msgs.size());
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 3.1\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-3.1", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 2, 3);

  run_producer("producer1, 0, 0x55, 1, BeginCommit, DoFlush",
               "producer1, 0, 0x10, 3, BeginOpen, DoFlush",
               "producer1, 1, 0x20, 3, ContinueOpen, DoFlush",
               "producer1, 0, 0x30, 3, ContinueAbort, DoFlush",
               "producer3, 0, 0x00, 1, None, DoFlush",
               "producer1, 1, 0x44, 1, BeginCommit, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 2,
              "Consumed unexpected number of messages. "
              "Expected 2, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 &&
                  0x55 == (unsigned char)msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[1]->key_len() >= 1 &&
                  0x00 == (unsigned char)msgs[1]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);
  msgs = consume_messages(c, topic_name, 1);
  TEST_ASSERT(msgs.size() == 1,
              "Consumed unexpected number of messages. "
              "Expected 1, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 &&
                  0x44 == (unsigned char)msgs[0]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 4 - simultaneous transactions (simple).\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-4", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer3, 0, 0x10, 1, None, DoFlush",
               "producer1, 0, 0x20, 3, BeginOpen, DoFlush",
               "producer2, 0, 0x30, 3, BeginOpen, DoFlush",
               "producer1, 0, 0x40, 3, ContinueCommit, DoFlush",
               "producer2, 0, 0x50, 3, ContinueAbort, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 7,
              "Consumed unexpected number of messages. "
              "Expected 7, got: %d",
              (int)msgs.size());
  delete_messages(msgs);
  c->close();
  delete c;

  c    = create_consumer(topic_name, "READ_UNCOMMITTED");
  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 13,
              "Consumed unexpected number of messages. "
              "Expected 13, got: %d",
              (int)msgs.size());
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 4.1\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-4.1", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer3, 0, 0x10, 1, None, DoFlush",
               "producer1, 0, 0x20, 3, BeginOpen, DoFlush",
               "producer2, 0, 0x30, 3, BeginOpen, DoFlush",
               "producer1, 0, 0x40, 3, ContinueAbort, DoFlush",
               "producer2, 0, 0x50, 3, ContinueCommit, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 7,
              "Consumed unexpected number of messages. "
              "Expected 7, got: %d",
              (int)msgs.size());
  delete_messages(msgs);
  c->close();
  delete c;

  c    = create_consumer(topic_name, "READ_UNCOMMITTED");
  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 13,
              "Consumed unexpected number of messages. "
              "Expected 13, got: %d",
              (int)msgs.size());
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 4.2\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-4.2", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer3, 0, 0x10, 1, None, DoFlush",
               "producer1, 0, 0x20, 3, BeginOpen, DoFlush",
               "producer2, 0, 0x30, 3, BeginOpen, DoFlush",
               "producer1, 0, 0x40, 3, ContinueCommit, DoFlush",
               "producer2, 0, 0x50, 3, ContinueCommit, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 13,
              "Consumed unexpected number of messages. "
              "Expected 7, got: %d",
              (int)msgs.size());
  delete_messages(msgs);
  c->close();
  delete c;

  c    = create_consumer(topic_name, "READ_UNCOMMITTED");
  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 13,
              "Consumed unexpected number of messages. "
              "Expected 13, got: %d",
              (int)msgs.size());
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 4.3\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-4.3", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer3, 0, 0x10, 1, None, DoFlush",
               "producer1, 0, 0x20, 3, BeginOpen, DoFlush",
               "producer2, 0, 0x30, 3, BeginOpen, DoFlush",
               "producer1, 0, 0x40, 3, ContinueAbort, DoFlush",
               "producer2, 0, 0x50, 3, ContinueAbort, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 1,
              "Consumed unexpected number of messages. "
              "Expected 7, got: %d",
              (int)msgs.size());
  delete_messages(msgs);
  c->close();
  delete c;

  c    = create_consumer(topic_name, "READ_UNCOMMITTED");
  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 13,
              "Consumed unexpected number of messages. "
              "Expected 13, got: %d",
              (int)msgs.size());
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;



  Test::Say(_C_BLU "Test 5 - split transaction across message sets.\n" _C_CLR);

test5:
  topic_name = Test::mk_topic_name("0098-consumer_txn-5", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);

  run_producer("producer1, 0, 0x10, 2, BeginOpen, DontFlush", "sleep,200",
               "producer1, 0, 0x20, 2, ContinueAbort, DontFlush",
               "producer1, 0, 0x30, 2, BeginOpen, DontFlush", "sleep,200",
               "producer1, 0, 0x40, 2, ContinueCommit, DontFlush",
               "producer1, 0, 0x50, 2, BeginOpen, DontFlush", "sleep,200",
               "producer1, 0, 0x60, 2, ContinueAbort, DontFlush",
               "producer1, 0, 0xa0, 2, BeginOpen, DontFlush", "sleep,200",
               "producer1, 0, 0xb0, 2, ContinueCommit, DontFlush",
               "producer3, 0, 0x70, 1, None, DoFlush");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 9,
              "Consumed unexpected number of messages. "
              "Expected 9, got: %d",
              (int)msgs.size());
  TEST_ASSERT(msgs[0]->key_len() >= 1 &&
                  0x30 == (unsigned char)msgs[0]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[1]->key_len() >= 1 &&
                  0x31 == (unsigned char)msgs[1]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[2]->key_len() >= 1 &&
                  0x40 == (unsigned char)msgs[2]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[3]->key_len() >= 1 &&
                  0x41 == (unsigned char)msgs[3]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[4]->key_len() >= 1 &&
                  0xa0 == (unsigned char)msgs[4]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[5]->key_len() >= 1 &&
                  0xa1 == (unsigned char)msgs[5]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[6]->key_len() >= 1 &&
                  0xb0 == (unsigned char)msgs[6]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[7]->key_len() >= 1 &&
                  0xb1 == (unsigned char)msgs[7]->key()->c_str()[0],
              "Unexpected key");
  TEST_ASSERT(msgs[8]->key_len() >= 1 &&
                  0x70 == (unsigned char)msgs[8]->key()->c_str()[0],
              "Unexpected key");
  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;


  Test::Say(_C_BLU "Test 6 - transaction left open\n" _C_CLR);

  topic_name = Test::mk_topic_name("0098-consumer_txn-0", 1);
  c          = create_consumer(topic_name, "READ_COMMITTED");
  Test::create_topic(c, topic_name.c_str(), 1, 3);
  TestEventCb::topic = topic_name;

  run_producer("producer3, 0, 0x10, 1, None, DoFlush",
               "producer1, 0, 0x20, 3, BeginOpen, DoFlush",
               // prevent abort control message from being written.
               "exit,0");

  msgs = consume_messages(c, topic_name, 0);
  TEST_ASSERT(msgs.size() == 1,
              "Consumed unexpected number of messages. "
              "Expected 1, got: %d",
              (int)msgs.size());

  TEST_ASSERT(TestEventCb::partition_0_ls_offset + 3 ==
                  TestEventCb::partition_0_hi_offset,
              "Expected hi_offset to be 3 greater than ls_offset "
              "but got hi_offset: %" PRId64 ", ls_offset: %" PRId64,
              TestEventCb::partition_0_hi_offset,
              TestEventCb::partition_0_ls_offset);

  delete_messages(msgs);

  Test::delete_topic(c, topic_name.c_str());

  c->close();
  delete c;
}
#endif


extern "C" {
int main_0098_consumer_txn(int argc, char **argv) {
  if (test_needs_auth()) {
    Test::Skip(
        "Authentication or security configuration "
        "required on client: not supported in "
        "Java transactional producer: skipping tests\n");
    return 0;
  }
#if WITH_RAPIDJSON
  do_test_consumer_txn_test(true /* with java producer */);
  do_test_consumer_txn_test(false /* with librdkafka producer */);
#else
  Test::Skip("RapidJSON >=1.1.0 not available\n");
#endif
  return 0;
}
}