text: /*
   * librdkafka - Apache Kafka C library
   *
   * Copyright (c) 2012-2013, 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 "test.h"
  
  static RD_UNUSED
  void print_toppar_list (const rd_kafka_topic_partition_list_t *list) {
          int i;
  
          TEST_SAY("List count: %d\n", list->cnt);
  
          for (i = 0 ; i < list->cnt ; i++) {
                  const rd_kafka_topic_partition_t *a = &list->elems[i];
  
                  TEST_SAY(" #%d/%d: "
                           "%s [%"PRId32"] @ %"PRId64": "
                           "(%"PRIusz") \"%*s\"\n",
                           i, list->cnt,
                           a->topic,
                           a->partition,
                           a->offset,
                           a->metadata_size,
                           (int)a->metadata_size,
                           (const char *)a->metadata);
          }
  }
  
  
  static void compare_toppar_lists (
          const rd_kafka_topic_partition_list_t *lista,
          const rd_kafka_topic_partition_list_t *listb) {
          int i;
  
          TEST_ASSERT(lista->cnt == listb->cnt,
                      "different list lengths: %d != %d",
                      lista->cnt, listb->cnt);
  
          for (i = 0 ; i < lista->cnt ; i++) {
                  const rd_kafka_topic_partition_t *a = &lista->elems[i];
                  const rd_kafka_topic_partition_t *b = &listb->elems[i];
  
                  if (a->offset != b->offset ||
                      a->metadata_size != b->metadata_size ||
                      memcmp(a->metadata, b->metadata, a->metadata_size))
                          TEST_FAIL_LATER(
                                  "Lists did not match at element %d/%d:\n"
                                  " a: %s [%"PRId32"] @ %"PRId64": "
                                  "(%"PRIusz") \"%*s\"\n"
                                  " b: %s [%"PRId32"] @ %"PRId64": "
                                  "(%"PRIusz") \"%*s\"",
                                  i, lista->cnt,
                                  a->topic,
                                  a->partition,
                                  a->offset,
                                  a->metadata_size,
                                  (int)a->metadata_size,
                                  (const char *)a->metadata,
                                  b->topic,
                                  b->partition,
                                  b->offset,
                                  b->metadata_size,
                                  (int)b->metadata_size,
                                  (const char *)b->metadata);
          }
  
          TEST_LATER_CHECK();
  }
  
  
  static int commit_cb_cnt = 0;
  
  static void offset_commit_cb (rd_kafka_t *rk,
                                rd_kafka_resp_err_t err,
                                rd_kafka_topic_partition_list_t *list,
                                void *opaque) {
          commit_cb_cnt++;
          TEST_ASSERT(!err, "offset_commit_cb failure: %s",
                      rd_kafka_err2str(err));
  }
  
  
  static void
  commit_metadata (const char *group_id,
                   const rd_kafka_topic_partition_list_t *toppar_to_commit) {
          rd_kafka_resp_err_t err;
          rd_kafka_t *rk;
          rd_kafka_conf_t *conf;
  
          test_conf_init(&conf, NULL, 20/*timeout*/);
  
          test_conf_set(conf, "group.id", group_id);
  
          rd_kafka_conf_set_offset_commit_cb(conf, offset_commit_cb);
  
          /* Create kafka instance */
          rk = test_create_handle(RD_KAFKA_CONSUMER, conf);
  
          TEST_SAY("Committing:\n");
          print_toppar_list(toppar_to_commit);
  
          err = rd_kafka_commit(rk, toppar_to_commit, 0);
          TEST_ASSERT(!err, "rd_kafka_commit failed: %s", rd_kafka_err2str(err));
  
          while (commit_cb_cnt == 0)
                  rd_kafka_poll(rk, 1000);
  
          rd_kafka_destroy(rk);
  }
  
  
  static void
  get_committed_metadata (const char *group_id,
                          const rd_kafka_topic_partition_list_t *toppar_to_check,
                          const rd_kafka_topic_partition_list_t *expected_toppar) {
          rd_kafka_resp_err_t err;
          rd_kafka_t *rk;
          rd_kafka_conf_t *conf;
          rd_kafka_topic_partition_list_t *committed_toppar;
  
          test_conf_init(&conf, NULL, 20/*timeout*/);
  
          test_conf_set(conf, "group.id", group_id);
  
          committed_toppar = rd_kafka_topic_partition_list_copy(toppar_to_check);
  
          /* Create kafka instance */
          rk = test_create_handle(RD_KAFKA_CONSUMER, conf);
  
          err = rd_kafka_committed(rk, committed_toppar, tmout_multip(5000));
          TEST_ASSERT(!err, "rd_kafka_committed failed: %s",
                      rd_kafka_err2str(err));
  
          compare_toppar_lists(committed_toppar, expected_toppar);
  
          rd_kafka_topic_partition_list_destroy(committed_toppar);
  
          rd_kafka_destroy(rk);
  }
  
  int main_0099_commit_metadata (int argc, char **argv) {
          rd_kafka_topic_partition_list_t *origin_toppar;
          rd_kafka_topic_partition_list_t *expected_toppar;
          const char *topic = test_mk_topic_name("0099-commit_metadata", 0);
          char group_id[16];
  
          test_conf_init(NULL, NULL, 20/*timeout*/);
  
          test_str_id_generate(group_id, sizeof(group_id));
  
          test_create_topic(NULL, topic, 1, 1);
  
          origin_toppar = rd_kafka_topic_partition_list_new(1);
  
          rd_kafka_topic_partition_list_add(origin_toppar, topic, 0);
  
          expected_toppar = rd_kafka_topic_partition_list_copy(origin_toppar);
  
          expected_toppar->elems[0].offset = 42;
          expected_toppar->elems[0].metadata = rd_strdup("Hello world!");
          expected_toppar->elems[0].metadata_size =
                  strlen(expected_toppar->elems[0].metadata);
  
          get_committed_metadata(group_id, origin_toppar, origin_toppar);
  
          commit_metadata(group_id, expected_toppar);
  
          get_committed_metadata(group_id, origin_toppar, expected_toppar);
  
          rd_kafka_topic_partition_list_destroy(origin_toppar);
          rd_kafka_topic_partition_list_destroy(expected_toppar);
  
          return 0;
  }
  
  
  