diff options
Diffstat (limited to 'fluent-bit/lib/librdkafka-2.1.0/tests/test.c')
| -rw-r--r-- | fluent-bit/lib/librdkafka-2.1.0/tests/test.c | 6960 |
1 files changed, 0 insertions, 6960 deletions
diff --git a/fluent-bit/lib/librdkafka-2.1.0/tests/test.c b/fluent-bit/lib/librdkafka-2.1.0/tests/test.c deleted file mode 100644 index 71180c8f4..000000000 --- a/fluent-bit/lib/librdkafka-2.1.0/tests/test.c +++ /dev/null @@ -1,6960 +0,0 @@ -/* - * 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. - */ - - -#define _CRT_RAND_S // rand_s() on MSVC -#include <stdarg.h> -#include "test.h" -#include <signal.h> -#include <stdlib.h> -#include <stdio.h> - -#ifdef _WIN32 -#include <direct.h> /* _getcwd */ -#else -#include <sys/wait.h> /* waitpid */ -#endif - -/* Typical include path would be <librdkafka/rdkafka.h>, but this program - * is built from within the librdkafka source tree and thus differs. */ -#include "rdkafka.h" - -int test_level = 2; -int test_seed = 0; - -char test_mode[64] = "bare"; -char test_scenario[64] = "default"; -static volatile sig_atomic_t test_exit = 0; -static char test_topic_prefix[128] = "rdkafkatest"; -static int test_topic_random = 0; -int tests_running_cnt = 0; -int test_concurrent_max = 5; -int test_assert_on_fail = 0; -double test_timeout_multiplier = 1.0; -static char *test_sql_cmd = NULL; -int test_session_timeout_ms = 6000; -int test_broker_version; -static const char *test_broker_version_str = "2.4.0.0"; -int test_flags = 0; -int test_neg_flags = TEST_F_KNOWN_ISSUE; -/* run delete-test-topics.sh between each test (when concurrent_max = 1) */ -static int test_delete_topics_between = 0; -static const char *test_git_version = "HEAD"; -static const char *test_sockem_conf = ""; -int test_on_ci = 0; /* Tests are being run on CI, be more forgiving - * with regards to timeouts, etc. */ -int test_quick = 0; /** Run tests quickly */ -int test_idempotent_producer = 0; -int test_rusage = 0; /**< Check resource usage */ -/**< CPU speed calibration for rusage threshold checks. - * >1.0: CPU is slower than base line system, - * <1.0: CPU is faster than base line system. */ -double test_rusage_cpu_calibration = 1.0; -static const char *tests_to_run = NULL; /* all */ -static const char *subtests_to_run = NULL; /* all */ -static const char *tests_to_skip = NULL; /* none */ -int test_write_report = 0; /**< Write test report file */ - -static int show_summary = 1; -static int test_summary(int do_lock); - -/** - * Protects shared state, such as tests[] - */ -mtx_t test_mtx; -cnd_t test_cnd; - -static const char *test_states[] = { - "DNS", "SKIPPED", "RUNNING", "PASSED", "FAILED", -}; - - - -#define _TEST_DECL(NAME) extern int main_##NAME(int, char **) -#define _TEST(NAME, FLAGS, ...) \ - { .name = #NAME, .mainfunc = main_##NAME, .flags = FLAGS, __VA_ARGS__ } - - -/** - * Declare all tests here - */ -_TEST_DECL(0000_unittests); -_TEST_DECL(0001_multiobj); -_TEST_DECL(0002_unkpart); -_TEST_DECL(0003_msgmaxsize); -_TEST_DECL(0004_conf); -_TEST_DECL(0005_order); -_TEST_DECL(0006_symbols); -_TEST_DECL(0007_autotopic); -_TEST_DECL(0008_reqacks); -_TEST_DECL(0009_mock_cluster); -_TEST_DECL(0011_produce_batch); -_TEST_DECL(0012_produce_consume); -_TEST_DECL(0013_null_msgs); -_TEST_DECL(0014_reconsume_191); -_TEST_DECL(0015_offsets_seek); -_TEST_DECL(0016_client_swname); -_TEST_DECL(0017_compression); -_TEST_DECL(0018_cgrp_term); -_TEST_DECL(0019_list_groups); -_TEST_DECL(0020_destroy_hang); -_TEST_DECL(0021_rkt_destroy); -_TEST_DECL(0022_consume_batch); -_TEST_DECL(0022_consume_batch_local); -_TEST_DECL(0025_timers); -_TEST_DECL(0026_consume_pause); -_TEST_DECL(0028_long_topicnames); -_TEST_DECL(0029_assign_offset); -_TEST_DECL(0030_offset_commit); -_TEST_DECL(0031_get_offsets); -_TEST_DECL(0033_regex_subscribe); -_TEST_DECL(0033_regex_subscribe_local); -_TEST_DECL(0034_offset_reset); -_TEST_DECL(0034_offset_reset_mock); -_TEST_DECL(0035_api_version); -_TEST_DECL(0036_partial_fetch); -_TEST_DECL(0037_destroy_hang_local); -_TEST_DECL(0038_performance); -_TEST_DECL(0039_event_dr); -_TEST_DECL(0039_event_log); -_TEST_DECL(0039_event); -_TEST_DECL(0040_io_event); -_TEST_DECL(0041_fetch_max_bytes); -_TEST_DECL(0042_many_topics); -_TEST_DECL(0043_no_connection); -_TEST_DECL(0044_partition_cnt); -_TEST_DECL(0045_subscribe_update); -_TEST_DECL(0045_subscribe_update_topic_remove); -_TEST_DECL(0045_subscribe_update_non_exist_and_partchange); -_TEST_DECL(0045_subscribe_update_mock); -_TEST_DECL(0046_rkt_cache); -_TEST_DECL(0047_partial_buf_tmout); -_TEST_DECL(0048_partitioner); -_TEST_DECL(0049_consume_conn_close); -_TEST_DECL(0050_subscribe_adds); -_TEST_DECL(0051_assign_adds); -_TEST_DECL(0052_msg_timestamps); -_TEST_DECL(0053_stats_timing); -_TEST_DECL(0053_stats); -_TEST_DECL(0054_offset_time); -_TEST_DECL(0055_producer_latency); -_TEST_DECL(0056_balanced_group_mt); -_TEST_DECL(0057_invalid_topic); -_TEST_DECL(0058_log); -_TEST_DECL(0059_bsearch); -_TEST_DECL(0060_op_prio); -_TEST_DECL(0061_consumer_lag); -_TEST_DECL(0062_stats_event); -_TEST_DECL(0063_clusterid); -_TEST_DECL(0064_interceptors); -_TEST_DECL(0065_yield); -_TEST_DECL(0066_plugins); -_TEST_DECL(0067_empty_topic); -_TEST_DECL(0068_produce_timeout); -_TEST_DECL(0069_consumer_add_parts); -_TEST_DECL(0070_null_empty); -_TEST_DECL(0072_headers_ut); -_TEST_DECL(0073_headers); -_TEST_DECL(0074_producev); -_TEST_DECL(0075_retry); -_TEST_DECL(0076_produce_retry); -_TEST_DECL(0077_compaction); -_TEST_DECL(0078_c_from_cpp); -_TEST_DECL(0079_fork); -_TEST_DECL(0080_admin_ut); -_TEST_DECL(0081_admin); -_TEST_DECL(0082_fetch_max_bytes); -_TEST_DECL(0083_cb_event); -_TEST_DECL(0084_destroy_flags_local); -_TEST_DECL(0084_destroy_flags); -_TEST_DECL(0085_headers); -_TEST_DECL(0086_purge_local); -_TEST_DECL(0086_purge_remote); -_TEST_DECL(0088_produce_metadata_timeout); -_TEST_DECL(0089_max_poll_interval); -_TEST_DECL(0090_idempotence); -_TEST_DECL(0091_max_poll_interval_timeout); -_TEST_DECL(0092_mixed_msgver); -_TEST_DECL(0093_holb_consumer); -_TEST_DECL(0094_idempotence_msg_timeout); -_TEST_DECL(0095_all_brokers_down); -_TEST_DECL(0097_ssl_verify); -_TEST_DECL(0097_ssl_verify_local); -_TEST_DECL(0098_consumer_txn); -_TEST_DECL(0099_commit_metadata); -_TEST_DECL(0100_thread_interceptors); -_TEST_DECL(0101_fetch_from_follower); -_TEST_DECL(0102_static_group_rebalance); -_TEST_DECL(0103_transactions_local); -_TEST_DECL(0103_transactions); -_TEST_DECL(0104_fetch_from_follower_mock); -_TEST_DECL(0105_transactions_mock); -_TEST_DECL(0106_cgrp_sess_timeout); -_TEST_DECL(0107_topic_recreate); -_TEST_DECL(0109_auto_create_topics); -_TEST_DECL(0110_batch_size); -_TEST_DECL(0111_delay_create_topics); -_TEST_DECL(0112_assign_unknown_part); -_TEST_DECL(0113_cooperative_rebalance_local); -_TEST_DECL(0113_cooperative_rebalance); -_TEST_DECL(0114_sticky_partitioning); -_TEST_DECL(0115_producer_auth); -_TEST_DECL(0116_kafkaconsumer_close); -_TEST_DECL(0117_mock_errors); -_TEST_DECL(0118_commit_rebalance); -_TEST_DECL(0119_consumer_auth); -_TEST_DECL(0120_asymmetric_subscription); -_TEST_DECL(0121_clusterid); -_TEST_DECL(0122_buffer_cleaning_after_rebalance); -_TEST_DECL(0123_connections_max_idle); -_TEST_DECL(0124_openssl_invalid_engine); -_TEST_DECL(0125_immediate_flush); -_TEST_DECL(0126_oauthbearer_oidc); -_TEST_DECL(0128_sasl_callback_queue); -_TEST_DECL(0129_fetch_aborted_msgs); -_TEST_DECL(0130_store_offsets); -_TEST_DECL(0131_connect_timeout); -_TEST_DECL(0132_strategy_ordering); -_TEST_DECL(0133_ssl_keys); -_TEST_DECL(0134_ssl_provider); -_TEST_DECL(0135_sasl_credentials); -_TEST_DECL(0136_resolve_cb); -_TEST_DECL(0137_barrier_batch_consume); -_TEST_DECL(0138_admin_mock); - -/* Manual tests */ -_TEST_DECL(8000_idle); - - -/* Define test resource usage thresholds if the default limits - * are not tolerable. - * - * Fields: - * .ucpu - Max User CPU percentage (double) - * .scpu - Max System/Kernel CPU percentage (double) - * .rss - Max RSS (memory) in megabytes (double) - * .ctxsw - Max number of voluntary context switches (int) - * - * Also see test_rusage_check_thresholds() in rusage.c - * - * Make a comment in the _THRES() below why the extra thresholds are required. - * - * Usage: - * _TEST(00...., ..., - * _THRES(.ucpu = 15.0)), <-- Max 15% User CPU usage - */ -#define _THRES(...) .rusage_thres = {__VA_ARGS__} - -/** - * Define all tests here - */ -struct test tests[] = { - /* Special MAIN test to hold over-all timings, etc. */ - {.name = "<MAIN>", .flags = TEST_F_LOCAL}, - _TEST(0000_unittests, - TEST_F_LOCAL, - /* The msgq insert order tests are heavy on - * user CPU (memory scan), RSS, and - * system CPU (lots of allocations -> madvise(2)). */ - _THRES(.ucpu = 100.0, .scpu = 20.0, .rss = 900.0)), - _TEST(0001_multiobj, 0), - _TEST(0002_unkpart, 0), - _TEST(0003_msgmaxsize, 0), - _TEST(0004_conf, TEST_F_LOCAL), - _TEST(0005_order, 0), - _TEST(0006_symbols, TEST_F_LOCAL), - _TEST(0007_autotopic, 0), - _TEST(0008_reqacks, 0), - _TEST(0009_mock_cluster, - TEST_F_LOCAL, - /* Mock cluster requires MsgVersion 2 */ - TEST_BRKVER(0, 11, 0, 0)), - _TEST(0011_produce_batch, - 0, - /* Produces a lot of messages */ - _THRES(.ucpu = 40.0, .scpu = 8.0)), - _TEST(0012_produce_consume, 0), - _TEST(0013_null_msgs, 0), - _TEST(0014_reconsume_191, 0), - _TEST(0015_offsets_seek, 0), - _TEST(0016_client_swname, 0), - _TEST(0017_compression, 0), - _TEST(0018_cgrp_term, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0019_list_groups, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0020_destroy_hang, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0021_rkt_destroy, 0), - _TEST(0022_consume_batch, 0), - _TEST(0022_consume_batch_local, TEST_F_LOCAL), - _TEST(0025_timers, TEST_F_LOCAL), - _TEST(0026_consume_pause, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0028_long_topicnames, - TEST_F_KNOWN_ISSUE, - TEST_BRKVER(0, 9, 0, 0), - .extra = "https://github.com/edenhill/librdkafka/issues/529"), - _TEST(0029_assign_offset, 0), - _TEST(0030_offset_commit, - 0, - TEST_BRKVER(0, 9, 0, 0), - /* Loops over committed() until timeout */ - _THRES(.ucpu = 10.0, .scpu = 5.0)), - _TEST(0031_get_offsets, 0), - _TEST(0033_regex_subscribe, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0033_regex_subscribe_local, TEST_F_LOCAL), - _TEST(0034_offset_reset, 0), - _TEST(0034_offset_reset_mock, TEST_F_LOCAL), - _TEST(0035_api_version, 0), - _TEST(0036_partial_fetch, 0), - _TEST(0037_destroy_hang_local, TEST_F_LOCAL), - _TEST(0038_performance, - 0, - /* Produces and consumes a lot of messages */ - _THRES(.ucpu = 150.0, .scpu = 10)), - _TEST(0039_event_dr, 0), - _TEST(0039_event_log, TEST_F_LOCAL), - _TEST(0039_event, TEST_F_LOCAL), - _TEST(0040_io_event, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0041_fetch_max_bytes, - 0, - /* Re-fetches large messages multiple times */ - _THRES(.ucpu = 20.0, .scpu = 10.0)), - _TEST(0042_many_topics, 0), - _TEST(0043_no_connection, TEST_F_LOCAL), - _TEST(0044_partition_cnt, - 0, - TEST_BRKVER(1, 0, 0, 0), - /* Produces a lot of messages */ - _THRES(.ucpu = 30.0)), - _TEST(0045_subscribe_update, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0045_subscribe_update_topic_remove, - 0, - TEST_BRKVER(0, 9, 0, 0), - .scenario = "noautocreate"), - _TEST(0045_subscribe_update_non_exist_and_partchange, - 0, - TEST_BRKVER(0, 9, 0, 0), - .scenario = "noautocreate"), - _TEST(0045_subscribe_update_mock, TEST_F_LOCAL), - _TEST(0046_rkt_cache, TEST_F_LOCAL), - _TEST(0047_partial_buf_tmout, TEST_F_KNOWN_ISSUE), - _TEST(0048_partitioner, - 0, - /* Produces many small messages */ - _THRES(.ucpu = 10.0, .scpu = 5.0)), -#if WITH_SOCKEM - _TEST(0049_consume_conn_close, TEST_F_SOCKEM, TEST_BRKVER(0, 9, 0, 0)), -#endif - _TEST(0050_subscribe_adds, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0051_assign_adds, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0052_msg_timestamps, 0, TEST_BRKVER(0, 10, 0, 0)), - _TEST(0053_stats_timing, TEST_F_LOCAL), - _TEST(0053_stats, 0), - _TEST(0054_offset_time, 0, TEST_BRKVER(0, 10, 1, 0)), - _TEST(0055_producer_latency, TEST_F_KNOWN_ISSUE_WIN32), - _TEST(0056_balanced_group_mt, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0057_invalid_topic, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0058_log, TEST_F_LOCAL), - _TEST(0059_bsearch, 0, TEST_BRKVER(0, 10, 0, 0)), - _TEST(0060_op_prio, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0061_consumer_lag, 0), - _TEST(0062_stats_event, TEST_F_LOCAL), - _TEST(0063_clusterid, 0, TEST_BRKVER(0, 10, 1, 0)), - _TEST(0064_interceptors, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0065_yield, 0), - _TEST(0066_plugins, - TEST_F_LOCAL | TEST_F_KNOWN_ISSUE_WIN32 | TEST_F_KNOWN_ISSUE_OSX, - .extra = - "dynamic loading of tests might not be fixed for this platform"), - _TEST(0067_empty_topic, 0), -#if WITH_SOCKEM - _TEST(0068_produce_timeout, TEST_F_SOCKEM), -#endif - _TEST(0069_consumer_add_parts, - TEST_F_KNOWN_ISSUE_WIN32, - TEST_BRKVER(1, 0, 0, 0)), - _TEST(0070_null_empty, 0), - _TEST(0072_headers_ut, TEST_F_LOCAL), - _TEST(0073_headers, 0, TEST_BRKVER(0, 11, 0, 0)), - _TEST(0074_producev, TEST_F_LOCAL), -#if WITH_SOCKEM - _TEST(0075_retry, TEST_F_SOCKEM), -#endif - _TEST(0076_produce_retry, TEST_F_SOCKEM), - _TEST(0077_compaction, - 0, - /* The test itself requires message headers */ - TEST_BRKVER(0, 11, 0, 0)), - _TEST(0078_c_from_cpp, TEST_F_LOCAL), - _TEST(0079_fork, - TEST_F_LOCAL | TEST_F_KNOWN_ISSUE, - .extra = "using a fork():ed rd_kafka_t is not supported and will " - "most likely hang"), - _TEST(0080_admin_ut, TEST_F_LOCAL), - _TEST(0081_admin, 0, TEST_BRKVER(0, 10, 2, 0)), - _TEST(0082_fetch_max_bytes, 0, TEST_BRKVER(0, 10, 1, 0)), - _TEST(0083_cb_event, 0, TEST_BRKVER(0, 9, 0, 0)), - _TEST(0084_destroy_flags_local, TEST_F_LOCAL), - _TEST(0084_destroy_flags, 0), - _TEST(0085_headers, 0, TEST_BRKVER(0, 11, 0, 0)), - _TEST(0086_purge_local, TEST_F_LOCAL), - _TEST(0086_purge_remote, 0), -#if WITH_SOCKEM - _TEST(0088_produce_metadata_timeout, TEST_F_SOCKEM), -#endif - _TEST(0089_max_poll_interval, 0, TEST_BRKVER(0, 10, 1, 0)), - _TEST(0090_idempotence, 0, TEST_BRKVER(0, 11, 0, 0)), - _TEST(0091_max_poll_interval_timeout, 0, TEST_BRKVER(0, 10, 1, 0)), - _TEST(0092_mixed_msgver, 0, TEST_BRKVER(0, 11, 0, 0)), - _TEST(0093_holb_consumer, 0, TEST_BRKVER(0, 10, 1, 0)), -#if WITH_SOCKEM - _TEST(0094_idempotence_msg_timeout, - TEST_F_SOCKEM, - TEST_BRKVER(0, 11, 0, 0)), -#endif - _TEST(0095_all_brokers_down, TEST_F_LOCAL), - _TEST(0097_ssl_verify, 0), - _TEST(0097_ssl_verify_local, TEST_F_LOCAL), - _TEST(0098_consumer_txn, 0, TEST_BRKVER(0, 11, 0, 0)), - _TEST(0099_commit_metadata, 0), - _TEST(0100_thread_interceptors, TEST_F_LOCAL), - _TEST(0101_fetch_from_follower, 0, TEST_BRKVER(2, 4, 0, 0)), - _TEST(0102_static_group_rebalance, 0, TEST_BRKVER(2, 3, 0, 0)), - _TEST(0103_transactions_local, TEST_F_LOCAL), - _TEST(0103_transactions, - 0, - TEST_BRKVER(0, 11, 0, 0), - .scenario = "default,ak23"), - _TEST(0104_fetch_from_follower_mock, TEST_F_LOCAL, TEST_BRKVER(2, 4, 0, 0)), - _TEST(0105_transactions_mock, TEST_F_LOCAL, TEST_BRKVER(0, 11, 0, 0)), - _TEST(0106_cgrp_sess_timeout, TEST_F_LOCAL, TEST_BRKVER(0, 11, 0, 0)), - _TEST(0107_topic_recreate, - 0, - TEST_BRKVER_TOPIC_ADMINAPI, - .scenario = "noautocreate"), - _TEST(0109_auto_create_topics, 0), - _TEST(0110_batch_size, 0), - _TEST(0111_delay_create_topics, - 0, - TEST_BRKVER_TOPIC_ADMINAPI, - .scenario = "noautocreate"), - _TEST(0112_assign_unknown_part, 0), - _TEST(0113_cooperative_rebalance_local, - TEST_F_LOCAL, - TEST_BRKVER(2, 4, 0, 0)), - _TEST(0113_cooperative_rebalance, 0, TEST_BRKVER(2, 4, 0, 0)), - _TEST(0114_sticky_partitioning, 0), - _TEST(0115_producer_auth, 0, TEST_BRKVER(2, 1, 0, 0)), - _TEST(0116_kafkaconsumer_close, TEST_F_LOCAL), - _TEST(0117_mock_errors, TEST_F_LOCAL), - _TEST(0118_commit_rebalance, 0), - _TEST(0119_consumer_auth, 0, TEST_BRKVER(2, 1, 0, 0)), - _TEST(0120_asymmetric_subscription, TEST_F_LOCAL), - _TEST(0121_clusterid, TEST_F_LOCAL), - _TEST(0122_buffer_cleaning_after_rebalance, 0, TEST_BRKVER(2, 4, 0, 0)), - _TEST(0123_connections_max_idle, 0), - _TEST(0124_openssl_invalid_engine, TEST_F_LOCAL), - _TEST(0125_immediate_flush, 0), - _TEST(0126_oauthbearer_oidc, 0, TEST_BRKVER(3, 1, 0, 0)), - _TEST(0128_sasl_callback_queue, TEST_F_LOCAL, TEST_BRKVER(2, 0, 0, 0)), - _TEST(0129_fetch_aborted_msgs, 0, TEST_BRKVER(0, 11, 0, 0)), - _TEST(0130_store_offsets, 0), - _TEST(0131_connect_timeout, TEST_F_LOCAL), - _TEST(0132_strategy_ordering, 0, TEST_BRKVER(2, 4, 0, 0)), - _TEST(0133_ssl_keys, TEST_F_LOCAL), - _TEST(0134_ssl_provider, TEST_F_LOCAL), - _TEST(0135_sasl_credentials, 0), - _TEST(0136_resolve_cb, TEST_F_LOCAL), - _TEST(0137_barrier_batch_consume, 0), - _TEST(0138_admin_mock, TEST_F_LOCAL, TEST_BRKVER(2, 4, 0, 0)), - - /* Manual tests */ - _TEST(8000_idle, TEST_F_MANUAL), - - {NULL}}; - - -RD_TLS struct test *test_curr = &tests[0]; - - - -#if WITH_SOCKEM -/** - * Socket network emulation with sockem - */ - -static void test_socket_add(struct test *test, sockem_t *skm) { - TEST_LOCK(); - rd_list_add(&test->sockets, skm); - TEST_UNLOCK(); -} - -static void test_socket_del(struct test *test, sockem_t *skm, int do_lock) { - if (do_lock) - TEST_LOCK(); - /* Best effort, skm might not have been added if connect_cb failed */ - rd_list_remove(&test->sockets, skm); - if (do_lock) - TEST_UNLOCK(); -} - -int test_socket_sockem_set_all(const char *key, int val) { - int i; - sockem_t *skm; - int cnt = 0; - - TEST_LOCK(); - - cnt = rd_list_cnt(&test_curr->sockets); - TEST_SAY("Setting sockem %s=%d on %s%d socket(s)\n", key, val, - cnt > 0 ? "" : _C_RED, cnt); - - RD_LIST_FOREACH(skm, &test_curr->sockets, i) { - if (sockem_set(skm, key, val, NULL) == -1) - TEST_FAIL("sockem_set(%s, %d) failed", key, val); - } - - TEST_UNLOCK(); - - return cnt; -} - -void test_socket_sockem_set(int s, const char *key, int value) { - sockem_t *skm; - - TEST_LOCK(); - skm = sockem_find(s); - if (skm) - sockem_set(skm, key, value, NULL); - TEST_UNLOCK(); -} - -void test_socket_close_all(struct test *test, int reinit) { - TEST_LOCK(); - rd_list_destroy(&test->sockets); - if (reinit) - rd_list_init(&test->sockets, 16, (void *)sockem_close); - TEST_UNLOCK(); -} - - -static int test_connect_cb(int s, - const struct sockaddr *addr, - int addrlen, - const char *id, - void *opaque) { - struct test *test = opaque; - sockem_t *skm; - int r; - - skm = sockem_connect(s, addr, addrlen, test_sockem_conf, 0, NULL); - if (!skm) - return errno; - - if (test->connect_cb) { - r = test->connect_cb(test, skm, id); - if (r) - return r; - } - - test_socket_add(test, skm); - - return 0; -} - -static int test_closesocket_cb(int s, void *opaque) { - struct test *test = opaque; - sockem_t *skm; - - TEST_LOCK(); - skm = sockem_find(s); - if (skm) { - /* Close sockem's sockets */ - sockem_close(skm); - test_socket_del(test, skm, 0 /*nolock*/); - } - TEST_UNLOCK(); - - /* Close librdkafka's socket */ -#ifdef _WIN32 - closesocket(s); -#else - close(s); -#endif - - return 0; -} - - -void test_socket_enable(rd_kafka_conf_t *conf) { - rd_kafka_conf_set_connect_cb(conf, test_connect_cb); - rd_kafka_conf_set_closesocket_cb(conf, test_closesocket_cb); - rd_kafka_conf_set_opaque(conf, test_curr); -} -#endif /* WITH_SOCKEM */ - -/** - * @brief For use as the is_fatal_cb(), treating no errors as test-fatal. - */ -int test_error_is_not_fatal_cb(rd_kafka_t *rk, - rd_kafka_resp_err_t err, - const char *reason) { - return 0; -} - -static void -test_error_cb(rd_kafka_t *rk, int err, const char *reason, void *opaque) { - if (test_curr->is_fatal_cb && - !test_curr->is_fatal_cb(rk, err, reason)) { - TEST_SAY(_C_YEL "%s rdkafka error (non-testfatal): %s: %s\n", - rd_kafka_name(rk), rd_kafka_err2str(err), reason); - } else { - if (err == RD_KAFKA_RESP_ERR__FATAL) { - char errstr[512]; - TEST_SAY(_C_RED "%s Fatal error: %s\n", - rd_kafka_name(rk), reason); - - err = rd_kafka_fatal_error(rk, errstr, sizeof(errstr)); - - if (test_curr->is_fatal_cb && - !test_curr->is_fatal_cb(rk, err, reason)) - TEST_SAY(_C_YEL - "%s rdkafka ignored FATAL error: " - "%s: %s\n", - rd_kafka_name(rk), - rd_kafka_err2str(err), errstr); - else - TEST_FAIL("%s rdkafka FATAL error: %s: %s", - rd_kafka_name(rk), - rd_kafka_err2str(err), errstr); - - } else { - TEST_FAIL("%s rdkafka error: %s: %s", rd_kafka_name(rk), - rd_kafka_err2str(err), reason); - } - } -} - -static int -test_stats_cb(rd_kafka_t *rk, char *json, size_t json_len, void *opaque) { - struct test *test = test_curr; - if (test->stats_fp) - fprintf(test->stats_fp, - "{\"test\": \"%s\", \"instance\":\"%s\", " - "\"stats\": %s}\n", - test->name, rd_kafka_name(rk), json); - return 0; -} - - -/** - * @brief Limit the test run time (in seconds) - */ -void test_timeout_set(int timeout) { - TEST_LOCK(); - TEST_SAY("Setting test timeout to %ds * %.1f\n", timeout, - test_timeout_multiplier); - timeout = (int)((double)timeout * test_timeout_multiplier); - test_curr->timeout = test_clock() + ((int64_t)timeout * 1000000); - TEST_UNLOCK(); -} - -int tmout_multip(int msecs) { - int r; - TEST_LOCK(); - r = (int)(((double)(msecs)) * test_timeout_multiplier); - TEST_UNLOCK(); - return r; -} - - - -#ifdef _WIN32 -static void test_init_win32(void) { - /* Enable VT emulation to support colored output. */ - HANDLE hOut = GetStdHandle(STD_OUTPUT_HANDLE); - DWORD dwMode = 0; - - if (hOut == INVALID_HANDLE_VALUE || !GetConsoleMode(hOut, &dwMode)) - return; - -#ifndef ENABLE_VIRTUAL_TERMINAL_PROCESSING -#define ENABLE_VIRTUAL_TERMINAL_PROCESSING 0x4 -#endif - dwMode |= ENABLE_VIRTUAL_TERMINAL_PROCESSING; - SetConsoleMode(hOut, dwMode); -} -#endif - - -static void test_init(void) { - int seed; - const char *tmp; - - - if (test_seed) - return; - - if ((tmp = test_getenv("TEST_LEVEL", NULL))) - test_level = atoi(tmp); - if ((tmp = test_getenv("TEST_MODE", NULL))) - strncpy(test_mode, tmp, sizeof(test_mode) - 1); - if ((tmp = test_getenv("TEST_SCENARIO", NULL))) - strncpy(test_scenario, tmp, sizeof(test_scenario) - 1); - if ((tmp = test_getenv("TEST_SOCKEM", NULL))) - test_sockem_conf = tmp; - if ((tmp = test_getenv("TEST_SEED", NULL))) - seed = atoi(tmp); - else - seed = test_clock() & 0xffffffff; - if ((tmp = test_getenv("TEST_CPU_CALIBRATION", NULL))) { - test_rusage_cpu_calibration = strtod(tmp, NULL); - if (test_rusage_cpu_calibration < 0.00001) { - fprintf(stderr, - "%% Invalid CPU calibration " - "value (from TEST_CPU_CALIBRATION env): %s\n", - tmp); - exit(1); - } - } - -#ifdef _WIN32 - test_init_win32(); - { - LARGE_INTEGER cycl; - QueryPerformanceCounter(&cycl); - seed = (int)cycl.QuadPart; - } -#endif - srand(seed); - test_seed = seed; -} - - -const char *test_mk_topic_name(const char *suffix, int randomized) { - static RD_TLS char ret[512]; - - /* Strip main_ prefix (caller is using __FUNCTION__) */ - if (!strncmp(suffix, "main_", 5)) - suffix += 5; - - if (test_topic_random || randomized) - rd_snprintf(ret, sizeof(ret), "%s_rnd%" PRIx64 "_%s", - test_topic_prefix, test_id_generate(), suffix); - else - rd_snprintf(ret, sizeof(ret), "%s_%s", test_topic_prefix, - suffix); - - TEST_SAY("Using topic \"%s\"\n", ret); - - return ret; -} - - -/** - * @brief Set special test config property - * @returns 1 if property was known, else 0. - */ -int test_set_special_conf(const char *name, const char *val, int *timeoutp) { - if (!strcmp(name, "test.timeout.multiplier")) { - TEST_LOCK(); - test_timeout_multiplier = strtod(val, NULL); - TEST_UNLOCK(); - *timeoutp = tmout_multip((*timeoutp) * 1000) / 1000; - } else if (!strcmp(name, "test.topic.prefix")) { - rd_snprintf(test_topic_prefix, sizeof(test_topic_prefix), "%s", - val); - } else if (!strcmp(name, "test.topic.random")) { - if (!strcmp(val, "true") || !strcmp(val, "1")) - test_topic_random = 1; - else - test_topic_random = 0; - } else if (!strcmp(name, "test.concurrent.max")) { - TEST_LOCK(); - test_concurrent_max = (int)strtod(val, NULL); - TEST_UNLOCK(); - } else if (!strcmp(name, "test.sql.command")) { - TEST_LOCK(); - if (test_sql_cmd) - rd_free(test_sql_cmd); - test_sql_cmd = rd_strdup(val); - TEST_UNLOCK(); - } else - return 0; - - return 1; -} - -static void test_read_conf_file(const char *conf_path, - rd_kafka_conf_t *conf, - rd_kafka_topic_conf_t *topic_conf, - int *timeoutp) { - FILE *fp; - char buf[1024]; - int line = 0; - -#ifndef _WIN32 - fp = fopen(conf_path, "r"); -#else - fp = NULL; - errno = fopen_s(&fp, conf_path, "r"); -#endif - if (!fp) { - if (errno == ENOENT) { - TEST_SAY("Test config file %s not found\n", conf_path); - return; - } else - TEST_FAIL("Failed to read %s: %s", conf_path, - strerror(errno)); - } - - while (fgets(buf, sizeof(buf) - 1, fp)) { - char *t; - char *b = buf; - rd_kafka_conf_res_t res = RD_KAFKA_CONF_UNKNOWN; - char *name, *val; - char errstr[512]; - - line++; - if ((t = strchr(b, '\n'))) - *t = '\0'; - - if (*b == '#' || !*b) - continue; - - if (!(t = strchr(b, '='))) - TEST_FAIL("%s:%i: expected name=value format\n", - conf_path, line); - - name = b; - *t = '\0'; - val = t + 1; - - if (test_set_special_conf(name, val, timeoutp)) - continue; - - if (!strncmp(name, "topic.", strlen("topic."))) { - name += strlen("topic."); - if (topic_conf) - res = rd_kafka_topic_conf_set(topic_conf, name, - val, errstr, - sizeof(errstr)); - else - res = RD_KAFKA_CONF_OK; - name -= strlen("topic."); - } - - if (res == RD_KAFKA_CONF_UNKNOWN) { - if (conf) - res = rd_kafka_conf_set(conf, name, val, errstr, - sizeof(errstr)); - else - res = RD_KAFKA_CONF_OK; - } - - if (res != RD_KAFKA_CONF_OK) - TEST_FAIL("%s:%i: %s\n", conf_path, line, errstr); - } - - fclose(fp); -} - -/** - * @brief Get path to test config file - */ -const char *test_conf_get_path(void) { - return test_getenv("RDKAFKA_TEST_CONF", "test.conf"); -} - -const char *test_getenv(const char *env, const char *def) { - return rd_getenv(env, def); -} - -void test_conf_common_init(rd_kafka_conf_t *conf, int timeout) { - if (conf) { - const char *tmp = test_getenv("TEST_DEBUG", NULL); - if (tmp) - test_conf_set(conf, "debug", tmp); - } - - if (timeout) - test_timeout_set(timeout); -} - - -/** - * Creates and sets up kafka configuration objects. - * Will read "test.conf" file if it exists. - */ -void test_conf_init(rd_kafka_conf_t **conf, - rd_kafka_topic_conf_t **topic_conf, - int timeout) { - const char *test_conf = test_conf_get_path(); - - if (conf) { - *conf = rd_kafka_conf_new(); - rd_kafka_conf_set(*conf, "client.id", test_curr->name, NULL, 0); - if (test_idempotent_producer) - test_conf_set(*conf, "enable.idempotence", "true"); - rd_kafka_conf_set_error_cb(*conf, test_error_cb); - rd_kafka_conf_set_stats_cb(*conf, test_stats_cb); - - /* Allow higher request timeouts on CI */ - if (test_on_ci) - test_conf_set(*conf, "request.timeout.ms", "10000"); - -#ifdef SIGIO - { - char buf[64]; - - /* Quick termination */ - rd_snprintf(buf, sizeof(buf), "%i", SIGIO); - rd_kafka_conf_set(*conf, "internal.termination.signal", - buf, NULL, 0); - signal(SIGIO, SIG_IGN); - } -#endif - } - -#if WITH_SOCKEM - if (*test_sockem_conf && conf) - test_socket_enable(*conf); -#endif - - if (topic_conf) - *topic_conf = rd_kafka_topic_conf_new(); - - /* Open and read optional local test configuration file, if any. */ - test_read_conf_file(test_conf, conf ? *conf : NULL, - topic_conf ? *topic_conf : NULL, &timeout); - - test_conf_common_init(conf ? *conf : NULL, timeout); -} - - -static RD_INLINE unsigned int test_rand(void) { - unsigned int r; -#ifdef _WIN32 - rand_s(&r); -#else - r = rand(); -#endif - return r; -} -/** - * Generate a "unique" test id. - */ -uint64_t test_id_generate(void) { - return (((uint64_t)test_rand()) << 32) | (uint64_t)test_rand(); -} - - -/** - * Generate a "unique" string id - */ -char *test_str_id_generate(char *dest, size_t dest_size) { - rd_snprintf(dest, dest_size, "%" PRId64, test_id_generate()); - return dest; -} - -/** - * Same as test_str_id_generate but returns a temporary string. - */ -const char *test_str_id_generate_tmp(void) { - static RD_TLS char ret[64]; - return test_str_id_generate(ret, sizeof(ret)); -} - -/** - * Format a message token. - * Pad's to dest_size. - */ -void test_msg_fmt(char *dest, - size_t dest_size, - uint64_t testid, - int32_t partition, - int msgid) { - size_t of; - - of = rd_snprintf(dest, dest_size, - "testid=%" PRIu64 ", partition=%" PRId32 ", msg=%i\n", - testid, partition, msgid); - if (of < dest_size - 1) { - memset(dest + of, '!', dest_size - of); - dest[dest_size - 1] = '\0'; - } -} - -/** - * @brief Prepare message value and key for test produce. - */ -void test_prepare_msg(uint64_t testid, - int32_t partition, - int msg_id, - char *val, - size_t val_size, - char *key, - size_t key_size) { - size_t of = 0; - - test_msg_fmt(key, key_size, testid, partition, msg_id); - - while (of < val_size) { - /* Copy-repeat key into val until val_size */ - size_t len = RD_MIN(val_size - of, key_size); - memcpy(val + of, key, len); - of += len; - } -} - - - -/** - * Parse a message token - */ -void test_msg_parse00(const char *func, - int line, - uint64_t testid, - int32_t exp_partition, - int *msgidp, - const char *topic, - int32_t partition, - int64_t offset, - const char *key, - size_t key_size) { - char buf[128]; - uint64_t in_testid; - int in_part; - - if (!key) - TEST_FAIL("%s:%i: Message (%s [%" PRId32 "] @ %" PRId64 - ") " - "has empty key\n", - func, line, topic, partition, offset); - - rd_snprintf(buf, sizeof(buf), "%.*s", (int)key_size, key); - - if (sscanf(buf, "testid=%" SCNu64 ", partition=%i, msg=%i\n", - &in_testid, &in_part, msgidp) != 3) - TEST_FAIL("%s:%i: Incorrect key format: %s", func, line, buf); - - - if (testid != in_testid || - (exp_partition != -1 && exp_partition != in_part)) - TEST_FAIL("%s:%i: Our testid %" PRIu64 - ", part %i did " - "not match message: \"%s\"\n", - func, line, testid, (int)exp_partition, buf); -} - -void test_msg_parse0(const char *func, - int line, - uint64_t testid, - rd_kafka_message_t *rkmessage, - int32_t exp_partition, - int *msgidp) { - test_msg_parse00(func, line, testid, exp_partition, msgidp, - rd_kafka_topic_name(rkmessage->rkt), - rkmessage->partition, rkmessage->offset, - (const char *)rkmessage->key, rkmessage->key_len); -} - - -struct run_args { - struct test *test; - int argc; - char **argv; -}; - -static int run_test0(struct run_args *run_args) { - struct test *test = run_args->test; - test_timing_t t_run; - int r; - char stats_file[256]; - - rd_snprintf(stats_file, sizeof(stats_file), "stats_%s_%" PRIu64 ".json", - test->name, test_id_generate()); - if (!(test->stats_fp = fopen(stats_file, "w+"))) - TEST_SAY("=== Failed to create stats file %s: %s ===\n", - stats_file, strerror(errno)); - - test_curr = test; - -#if WITH_SOCKEM - rd_list_init(&test->sockets, 16, (void *)sockem_close); -#endif - /* Don't check message status by default */ - test->exp_dr_status = (rd_kafka_msg_status_t)-1; - - TEST_SAY("================= Running test %s =================\n", - test->name); - if (test->stats_fp) - TEST_SAY("==== Stats written to file %s ====\n", stats_file); - - test_rusage_start(test_curr); - TIMING_START(&t_run, "%s", test->name); - test->start = t_run.ts_start; - - /* Run test main function */ - r = test->mainfunc(run_args->argc, run_args->argv); - - TIMING_STOP(&t_run); - test_rusage_stop(test_curr, - (double)TIMING_DURATION(&t_run) / 1000000.0); - - TEST_LOCK(); - test->duration = TIMING_DURATION(&t_run); - - if (test->state == TEST_SKIPPED) { - TEST_SAY( - "================= Test %s SKIPPED " - "=================\n", - run_args->test->name); - } else if (r) { - test->state = TEST_FAILED; - TEST_SAY( - "\033[31m" - "================= Test %s FAILED =================" - "\033[0m\n", - run_args->test->name); - } else { - test->state = TEST_PASSED; - TEST_SAY( - "\033[32m" - "================= Test %s PASSED =================" - "\033[0m\n", - run_args->test->name); - } - TEST_UNLOCK(); - - cnd_broadcast(&test_cnd); - -#if WITH_SOCKEM - test_socket_close_all(test, 0); -#endif - - if (test->stats_fp) { - long pos = ftell(test->stats_fp); - fclose(test->stats_fp); - test->stats_fp = NULL; - /* Delete file if nothing was written */ - if (pos == 0) { -#ifndef _WIN32 - unlink(stats_file); -#else - _unlink(stats_file); -#endif - } - } - - if (test_delete_topics_between && test_concurrent_max == 1) - test_delete_all_test_topics(60 * 1000); - - return r; -} - - - -static int run_test_from_thread(void *arg) { - struct run_args *run_args = arg; - - thrd_detach(thrd_current()); - - run_test0(run_args); - - TEST_LOCK(); - tests_running_cnt--; - TEST_UNLOCK(); - - free(run_args); - - return 0; -} - - -/** - * @brief Check running tests for timeouts. - * @locks TEST_LOCK MUST be held - */ -static void check_test_timeouts(void) { - int64_t now = test_clock(); - struct test *test; - - for (test = tests; test->name; test++) { - if (test->state != TEST_RUNNING) - continue; - - /* Timeout check */ - if (now > test->timeout) { - struct test *save_test = test_curr; - test_curr = test; - test->state = TEST_FAILED; - test_summary(0 /*no-locks*/); - TEST_FAIL0( - __FILE__, __LINE__, 0 /*nolock*/, 0 /*fail-later*/, - "Test %s%s%s%s timed out " - "(timeout set to %d seconds)\n", - test->name, *test->subtest ? " (" : "", - test->subtest, *test->subtest ? ")" : "", - (int)(test->timeout - test->start) / 1000000); - test_curr = save_test; - tests_running_cnt--; /* fail-later misses this*/ -#ifdef _WIN32 - TerminateThread(test->thrd, -1); -#else - pthread_kill(test->thrd, SIGKILL); -#endif - } - } -} - - -static int run_test(struct test *test, int argc, char **argv) { - struct run_args *run_args = calloc(1, sizeof(*run_args)); - int wait_cnt = 0; - - run_args->test = test; - run_args->argc = argc; - run_args->argv = argv; - - TEST_LOCK(); - while (tests_running_cnt >= test_concurrent_max) { - if (!(wait_cnt++ % 100)) - TEST_SAY( - "Too many tests running (%d >= %d): " - "postponing %s start...\n", - tests_running_cnt, test_concurrent_max, test->name); - cnd_timedwait_ms(&test_cnd, &test_mtx, 100); - - check_test_timeouts(); - } - tests_running_cnt++; - test->timeout = test_clock() + - (int64_t)(30.0 * 1000000.0 * test_timeout_multiplier); - test->state = TEST_RUNNING; - TEST_UNLOCK(); - - if (thrd_create(&test->thrd, run_test_from_thread, run_args) != - thrd_success) { - TEST_LOCK(); - tests_running_cnt--; - test->state = TEST_FAILED; - TEST_UNLOCK(); - - TEST_FAIL("Failed to start thread for test %s\n", test->name); - } - - return 0; -} - -static void run_tests(int argc, char **argv) { - struct test *test; - - for (test = tests; test->name; test++) { - char testnum[128]; - char *t; - const char *skip_reason = NULL; - rd_bool_t skip_silent = rd_false; - char tmp[128]; - const char *scenario = - test->scenario ? test->scenario : "default"; - - if (!test->mainfunc) - continue; - - /* Extract test number, as string */ - strncpy(testnum, test->name, sizeof(testnum) - 1); - testnum[sizeof(testnum) - 1] = '\0'; - if ((t = strchr(testnum, '_'))) - *t = '\0'; - - if ((test_flags && (test_flags & test->flags) != test_flags)) { - skip_reason = "filtered due to test flags"; - skip_silent = rd_true; - } - if ((test_neg_flags & ~test_flags) & test->flags) - skip_reason = "Filtered due to negative test flags"; - if (test_broker_version && - (test->minver > test_broker_version || - (test->maxver && test->maxver < test_broker_version))) { - rd_snprintf(tmp, sizeof(tmp), - "not applicable for broker " - "version %d.%d.%d.%d", - TEST_BRKVER_X(test_broker_version, 0), - TEST_BRKVER_X(test_broker_version, 1), - TEST_BRKVER_X(test_broker_version, 2), - TEST_BRKVER_X(test_broker_version, 3)); - skip_reason = tmp; - } - - if (!strstr(scenario, test_scenario)) { - rd_snprintf(tmp, sizeof(tmp), - "requires test scenario %s", scenario); - skip_silent = rd_true; - skip_reason = tmp; - } - - if (tests_to_run && !strstr(tests_to_run, testnum)) { - skip_reason = "not included in TESTS list"; - skip_silent = rd_true; - } else if (!tests_to_run && (test->flags & TEST_F_MANUAL)) { - skip_reason = "manual test"; - skip_silent = rd_true; - } else if (tests_to_skip && strstr(tests_to_skip, testnum)) - skip_reason = "included in TESTS_SKIP list"; - - if (!skip_reason) { - run_test(test, argc, argv); - } else { - if (skip_silent) { - TEST_SAYL(3, - "================= Skipping test %s " - "(%s) ================\n", - test->name, skip_reason); - TEST_LOCK(); - test->state = TEST_SKIPPED; - TEST_UNLOCK(); - } else { - test_curr = test; - TEST_SKIP("%s\n", skip_reason); - test_curr = &tests[0]; - } - } - } -} - -/** - * @brief Print summary for all tests. - * - * @returns the number of failed tests. - */ -static int test_summary(int do_lock) { - struct test *test; - FILE *report_fp = NULL; - char report_path[128]; - time_t t; - struct tm *tm; - char datestr[64]; - int64_t total_duration = 0; - int tests_run = 0; - int tests_failed = 0; - int tests_failed_known = 0; - int tests_passed = 0; - FILE *sql_fp = NULL; - const char *tmp; - - t = time(NULL); - tm = localtime(&t); - strftime(datestr, sizeof(datestr), "%Y%m%d%H%M%S", tm); - - if ((tmp = test_getenv("TEST_REPORT", NULL))) - rd_snprintf(report_path, sizeof(report_path), "%s", tmp); - else if (test_write_report) - rd_snprintf(report_path, sizeof(report_path), - "test_report_%s.json", datestr); - else - report_path[0] = '\0'; - - if (*report_path) { - report_fp = fopen(report_path, "w+"); - if (!report_fp) - TEST_WARN("Failed to create report file %s: %s\n", - report_path, strerror(errno)); - else - fprintf(report_fp, - "{ \"id\": \"%s_%s\", \"mode\": \"%s\", " - "\"scenario\": \"%s\", " - "\"date\": \"%s\", " - "\"git_version\": \"%s\", " - "\"broker_version\": \"%s\", " - "\"tests\": {", - datestr, test_mode, test_mode, test_scenario, - datestr, test_git_version, - test_broker_version_str); - } - - if (do_lock) - TEST_LOCK(); - - if (test_sql_cmd) { -#ifdef _WIN32 - sql_fp = _popen(test_sql_cmd, "w"); -#else - sql_fp = popen(test_sql_cmd, "w"); -#endif - if (!sql_fp) - TEST_WARN("Failed to execute test.sql.command: %s", - test_sql_cmd); - else - fprintf(sql_fp, - "CREATE TABLE IF NOT EXISTS " - "runs(runid text PRIMARY KEY, mode text, " - "date datetime, cnt int, passed int, " - "failed int, duration numeric);\n" - "CREATE TABLE IF NOT EXISTS " - "tests(runid text, mode text, name text, " - "state text, extra text, duration numeric);\n"); - } - - if (show_summary) - printf( - "TEST %s (%s, scenario %s) SUMMARY\n" - "#=========================================================" - "=========#\n", - datestr, test_mode, test_scenario); - - for (test = tests; test->name; test++) { - const char *color; - int64_t duration; - char extra[128] = ""; - int do_count = 1; - - if (!(duration = test->duration) && test->start > 0) - duration = test_clock() - test->start; - - if (test == tests) { - /* <MAIN> test: - * test accounts for total runtime. - * dont include in passed/run/failed counts. */ - total_duration = duration; - do_count = 0; - } - - switch (test->state) { - case TEST_PASSED: - color = _C_GRN; - if (do_count) { - tests_passed++; - tests_run++; - } - break; - case TEST_FAILED: - if (test->flags & TEST_F_KNOWN_ISSUE) { - rd_snprintf(extra, sizeof(extra), - " <-- known issue%s%s", - test->extra ? ": " : "", - test->extra ? test->extra : ""); - if (do_count) - tests_failed_known++; - } - color = _C_RED; - if (do_count) { - tests_failed++; - tests_run++; - } - break; - case TEST_RUNNING: - color = _C_MAG; - if (do_count) { - tests_failed++; /* All tests should be finished - */ - tests_run++; - } - break; - case TEST_NOT_STARTED: - color = _C_YEL; - if (test->extra) - rd_snprintf(extra, sizeof(extra), " %s", - test->extra); - break; - default: - color = _C_CYA; - break; - } - - if (show_summary && - (test->state != TEST_SKIPPED || *test->failstr || - (tests_to_run && !strncmp(tests_to_run, test->name, - strlen(tests_to_run))))) { - printf("|%s %-40s | %10s | %7.3fs %s|", color, - test->name, test_states[test->state], - (double)duration / 1000000.0, _C_CLR); - if (test->state == TEST_FAILED) - printf(_C_RED " %s" _C_CLR, test->failstr); - else if (test->state == TEST_SKIPPED) - printf(_C_CYA " %s" _C_CLR, test->failstr); - printf("%s\n", extra); - } - - if (report_fp) { - int i; - fprintf(report_fp, - "%s\"%s\": {" - "\"name\": \"%s\", " - "\"state\": \"%s\", " - "\"known_issue\": %s, " - "\"extra\": \"%s\", " - "\"duration\": %.3f, " - "\"report\": [ ", - test == tests ? "" : ", ", test->name, - test->name, test_states[test->state], - test->flags & TEST_F_KNOWN_ISSUE ? "true" - : "false", - test->extra ? test->extra : "", - (double)duration / 1000000.0); - - for (i = 0; i < test->report_cnt; i++) { - fprintf(report_fp, "%s%s ", i == 0 ? "" : ",", - test->report_arr[i]); - } - - fprintf(report_fp, "] }"); - } - - if (sql_fp) - fprintf(sql_fp, - "INSERT INTO tests VALUES(" - "'%s_%s', '%s', '%s', '%s', '%s', %f);\n", - datestr, test_mode, test_mode, test->name, - test_states[test->state], - test->extra ? test->extra : "", - (double)duration / 1000000.0); - } - if (do_lock) - TEST_UNLOCK(); - - if (show_summary) - printf( - "#=========================================================" - "=========#\n"); - - if (report_fp) { - fprintf(report_fp, - "}, " - "\"tests_run\": %d, " - "\"tests_passed\": %d, " - "\"tests_failed\": %d, " - "\"duration\": %.3f" - "}\n", - tests_run, tests_passed, tests_failed, - (double)total_duration / 1000000.0); - - fclose(report_fp); - TEST_SAY("# Test report written to %s\n", report_path); - } - - if (sql_fp) { - fprintf(sql_fp, - "INSERT INTO runs VALUES('%s_%s', '%s', datetime(), " - "%d, %d, %d, %f);\n", - datestr, test_mode, test_mode, tests_run, tests_passed, - tests_failed, (double)total_duration / 1000000.0); - fclose(sql_fp); - } - - return tests_failed - tests_failed_known; -} - -#ifndef _WIN32 -static void test_sig_term(int sig) { - if (test_exit) - exit(1); - fprintf(stderr, - "Exiting tests, waiting for running tests to finish.\n"); - test_exit = 1; -} -#endif - -/** - * Wait 'timeout' seconds for rdkafka to kill all its threads and clean up. - */ -static void test_wait_exit(int timeout) { - int r; - time_t start = time(NULL); - - while ((r = rd_kafka_thread_cnt()) && timeout-- >= 0) { - TEST_SAY("%i thread(s) in use by librdkafka, waiting...\n", r); - rd_sleep(1); - } - - TEST_SAY("%i thread(s) in use by librdkafka\n", r); - - if (r > 0) - TEST_FAIL("%i thread(s) still active in librdkafka", r); - - timeout -= (int)(time(NULL) - start); - if (timeout > 0) { - TEST_SAY( - "Waiting %d seconds for all librdkafka memory " - "to be released\n", - timeout); - if (rd_kafka_wait_destroyed(timeout * 1000) == -1) - TEST_FAIL( - "Not all internal librdkafka " - "objects destroyed\n"); - } -} - - - -/** - * @brief Test framework cleanup before termination. - */ -static void test_cleanup(void) { - struct test *test; - - /* Free report arrays */ - for (test = tests; test->name; test++) { - int i; - if (!test->report_arr) - continue; - for (i = 0; i < test->report_cnt; i++) - rd_free(test->report_arr[i]); - rd_free(test->report_arr); - test->report_arr = NULL; - } - - if (test_sql_cmd) - rd_free(test_sql_cmd); -} - - -int main(int argc, char **argv) { - int i, r; - test_timing_t t_all; - int a, b, c, d; - const char *tmpver; - - mtx_init(&test_mtx, mtx_plain); - cnd_init(&test_cnd); - - test_init(); - -#ifndef _WIN32 - signal(SIGINT, test_sig_term); -#endif - tests_to_run = test_getenv("TESTS", NULL); - subtests_to_run = test_getenv("SUBTESTS", NULL); - tests_to_skip = test_getenv("TESTS_SKIP", NULL); - tmpver = test_getenv("TEST_KAFKA_VERSION", NULL); - if (!tmpver) - tmpver = test_getenv("KAFKA_VERSION", test_broker_version_str); - test_broker_version_str = tmpver; - - test_git_version = test_getenv("RDKAFKA_GITVER", "HEAD"); - - /* Are we running on CI? */ - if (test_getenv("CI", NULL)) { - test_on_ci = 1; - test_concurrent_max = 3; - } - - test_conf_init(NULL, NULL, 10); - - for (i = 1; i < argc; i++) { - if (!strncmp(argv[i], "-p", 2) && strlen(argv[i]) > 2) { - if (test_rusage) { - fprintf(stderr, - "%% %s ignored: -R takes preceedence\n", - argv[i]); - continue; - } - test_concurrent_max = (int)strtod(argv[i] + 2, NULL); - } else if (!strcmp(argv[i], "-l")) - test_flags |= TEST_F_LOCAL; - else if (!strcmp(argv[i], "-L")) - test_neg_flags |= TEST_F_LOCAL; - else if (!strcmp(argv[i], "-a")) - test_assert_on_fail = 1; - else if (!strcmp(argv[i], "-k")) - test_flags |= TEST_F_KNOWN_ISSUE; - else if (!strcmp(argv[i], "-K")) - test_neg_flags |= TEST_F_KNOWN_ISSUE; - else if (!strcmp(argv[i], "-E")) - test_neg_flags |= TEST_F_SOCKEM; - else if (!strcmp(argv[i], "-V") && i + 1 < argc) - test_broker_version_str = argv[++i]; - else if (!strcmp(argv[i], "-s") && i + 1 < argc) - strncpy(test_scenario, argv[++i], - sizeof(test_scenario) - 1); - else if (!strcmp(argv[i], "-S")) - show_summary = 0; - else if (!strcmp(argv[i], "-D")) - test_delete_topics_between = 1; - else if (!strcmp(argv[i], "-P")) - test_idempotent_producer = 1; - else if (!strcmp(argv[i], "-Q")) - test_quick = 1; - else if (!strcmp(argv[i], "-r")) - test_write_report = 1; - else if (!strncmp(argv[i], "-R", 2)) { - test_rusage = 1; - test_concurrent_max = 1; - if (strlen(argv[i]) > strlen("-R")) { - test_rusage_cpu_calibration = - strtod(argv[i] + 2, NULL); - if (test_rusage_cpu_calibration < 0.00001) { - fprintf(stderr, - "%% Invalid CPU calibration " - "value: %s\n", - argv[i] + 2); - exit(1); - } - } - } else if (*argv[i] != '-') - tests_to_run = argv[i]; - else { - printf( - "Unknown option: %s\n" - "\n" - "Usage: %s [options] [<test-match-substr>]\n" - "Options:\n" - " -p<N> Run N tests in parallel\n" - " -l/-L Only/dont run local tests (no broker " - "needed)\n" - " -k/-K Only/dont run tests with known issues\n" - " -E Don't run sockem tests\n" - " -a Assert on failures\n" - " -r Write test_report_...json file.\n" - " -S Dont show test summary\n" - " -s <scenario> Test scenario.\n" - " -V <N.N.N.N> Broker version.\n" - " -D Delete all test topics between each test " - "(-p1) or after all tests\n" - " -P Run all tests with " - "`enable.idempotency=true`\n" - " -Q Run tests in quick mode: faster tests, " - "fewer iterations, less data.\n" - " -R Check resource usage thresholds.\n" - " -R<C> Check resource usage thresholds but " - "adjust CPU thresholds by C (float):\n" - " C < 1.0: CPU is faster than base line " - "system.\n" - " C > 1.0: CPU is slower than base line " - "system.\n" - " E.g. -R2.5 = CPU is 2.5x slower than " - "base line system.\n" - "\n" - "Environment variables:\n" - " TESTS - substring matched test to run (e.g., " - "0033)\n" - " SUBTESTS - substring matched subtest to run " - "(e.g., n_wildcard)\n" - " TEST_KAFKA_VERSION - broker version (e.g., " - "0.9.0.1)\n" - " TEST_SCENARIO - Test scenario\n" - " TEST_LEVEL - Test verbosity level\n" - " TEST_MODE - bare, helgrind, valgrind\n" - " TEST_SEED - random seed\n" - " RDKAFKA_TEST_CONF - test config file " - "(test.conf)\n" - " KAFKA_PATH - Path to kafka source dir\n" - " ZK_ADDRESS - Zookeeper address\n" - "\n", - argv[i], argv[0]); - exit(1); - } - } - - TEST_SAY("Git version: %s\n", test_git_version); - - if (!strcmp(test_broker_version_str, "trunk")) - test_broker_version_str = "9.9.9.9"; /* for now */ - - d = 0; - if (sscanf(test_broker_version_str, "%d.%d.%d.%d", &a, &b, &c, &d) < - 3) { - printf( - "%% Expected broker version to be in format " - "N.N.N (N=int), not %s\n", - test_broker_version_str); - exit(1); - } - test_broker_version = TEST_BRKVER(a, b, c, d); - TEST_SAY("Broker version: %s (%d.%d.%d.%d)\n", test_broker_version_str, - TEST_BRKVER_X(test_broker_version, 0), - TEST_BRKVER_X(test_broker_version, 1), - TEST_BRKVER_X(test_broker_version, 2), - TEST_BRKVER_X(test_broker_version, 3)); - - /* Set up fake "<MAIN>" test for all operations performed in - * the main thread rather than the per-test threads. - * Nice side effect is that we get timing and status for main as well.*/ - test_curr = &tests[0]; - test_curr->state = TEST_PASSED; - test_curr->start = test_clock(); - - if (test_on_ci) { - TEST_LOCK(); - test_timeout_multiplier += 2; - TEST_UNLOCK(); - } - - if (!strcmp(test_mode, "helgrind") || !strcmp(test_mode, "drd")) { - TEST_LOCK(); - test_timeout_multiplier += 5; - TEST_UNLOCK(); - } else if (!strcmp(test_mode, "valgrind")) { - TEST_LOCK(); - test_timeout_multiplier += 3; - TEST_UNLOCK(); - } - - /* Broker version 0.9 and api.version.request=true (which is default) - * will cause a 10s stall per connection. Instead of fixing - * that for each affected API in every test we increase the timeout - * multiplier accordingly instead. The typical consume timeout is 5 - * seconds, so a multiplier of 3 should be good. */ - if ((test_broker_version & 0xffff0000) == 0x00090000) - test_timeout_multiplier += 3; - - if (test_concurrent_max > 1) - test_timeout_multiplier += (double)test_concurrent_max / 3; - - TEST_SAY("Tests to run : %s\n", tests_to_run ? tests_to_run : "all"); - if (subtests_to_run) - TEST_SAY("Sub tests : %s\n", subtests_to_run); - if (tests_to_skip) - TEST_SAY("Skip tests : %s\n", tests_to_skip); - TEST_SAY("Test mode : %s%s%s\n", test_quick ? "quick, " : "", - test_mode, test_on_ci ? ", CI" : ""); - TEST_SAY("Test scenario: %s\n", test_scenario); - TEST_SAY("Test filter : %s\n", (test_flags & TEST_F_LOCAL) - ? "local tests only" - : "no filter"); - TEST_SAY("Test timeout multiplier: %.1f\n", test_timeout_multiplier); - TEST_SAY("Action on test failure: %s\n", - test_assert_on_fail ? "assert crash" : "continue other tests"); - if (test_rusage) - TEST_SAY("Test rusage : yes (%.2fx CPU calibration)\n", - test_rusage_cpu_calibration); - if (test_idempotent_producer) - TEST_SAY("Test Idempotent Producer: enabled\n"); - - { - char cwd[512], *pcwd; -#ifdef _WIN32 - pcwd = _getcwd(cwd, sizeof(cwd) - 1); -#else - pcwd = getcwd(cwd, sizeof(cwd) - 1); -#endif - if (pcwd) - TEST_SAY("Current directory: %s\n", cwd); - } - - test_timeout_set(30); - - TIMING_START(&t_all, "ALL-TESTS"); - - /* Run tests */ - run_tests(argc, argv); - - TEST_LOCK(); - while (tests_running_cnt > 0 && !test_exit) { - struct test *test; - - if (!test_quick && test_level >= 2) { - TEST_SAY("%d test(s) running:", tests_running_cnt); - - for (test = tests; test->name; test++) { - if (test->state != TEST_RUNNING) - continue; - - TEST_SAY0(" %s", test->name); - } - - TEST_SAY0("\n"); - } - - check_test_timeouts(); - - TEST_UNLOCK(); - - if (test_quick) - rd_usleep(200 * 1000, NULL); - else - rd_sleep(1); - TEST_LOCK(); - } - - TIMING_STOP(&t_all); - - test_curr = &tests[0]; - test_curr->duration = test_clock() - test_curr->start; - - TEST_UNLOCK(); - - if (test_delete_topics_between) - test_delete_all_test_topics(60 * 1000); - - r = test_summary(1 /*lock*/) ? 1 : 0; - - /* Wait for everything to be cleaned up since broker destroys are - * handled in its own thread. */ - test_wait_exit(0); - - /* If we havent failed at this point then - * there were no threads leaked */ - if (r == 0) - TEST_SAY("\n============== ALL TESTS PASSED ==============\n"); - - test_cleanup(); - - if (r > 0) - TEST_FAIL("%d test(s) failed, see previous errors", r); - - return r; -} - - - -/****************************************************************************** - * - * Helpers - * - ******************************************************************************/ - -void test_dr_msg_cb(rd_kafka_t *rk, - const rd_kafka_message_t *rkmessage, - void *opaque) { - int *remainsp = rkmessage->_private; - static const char *status_names[] = { - [RD_KAFKA_MSG_STATUS_NOT_PERSISTED] = "NotPersisted", - [RD_KAFKA_MSG_STATUS_POSSIBLY_PERSISTED] = "PossiblyPersisted", - [RD_KAFKA_MSG_STATUS_PERSISTED] = "Persisted"}; - - TEST_SAYL(4, - "Delivery report: %s (%s) to %s [%" PRId32 - "] " - "at offset %" PRId64 " latency %.2fms\n", - rd_kafka_err2str(rkmessage->err), - status_names[rd_kafka_message_status(rkmessage)], - rd_kafka_topic_name(rkmessage->rkt), rkmessage->partition, - rkmessage->offset, - (float)rd_kafka_message_latency(rkmessage) / 1000.0); - - if (!test_curr->produce_sync) { - if (!test_curr->ignore_dr_err && - rkmessage->err != test_curr->exp_dr_err) - TEST_FAIL("Message delivery (to %s [%" PRId32 - "]) " - "failed: expected %s, got %s", - rd_kafka_topic_name(rkmessage->rkt), - rkmessage->partition, - rd_kafka_err2str(test_curr->exp_dr_err), - rd_kafka_err2str(rkmessage->err)); - - if ((int)test_curr->exp_dr_status != -1) { - rd_kafka_msg_status_t status = - rd_kafka_message_status(rkmessage); - - TEST_ASSERT(status == test_curr->exp_dr_status, - "Expected message status %s, not %s", - status_names[test_curr->exp_dr_status], - status_names[status]); - } - - /* Add message to msgver */ - if (!rkmessage->err && test_curr->dr_mv) - test_msgver_add_msg(rk, test_curr->dr_mv, rkmessage); - } - - if (remainsp) { - TEST_ASSERT(*remainsp > 0, - "Too many messages delivered (remains %i)", - *remainsp); - - (*remainsp)--; - } - - if (test_curr->produce_sync) - test_curr->produce_sync_err = rkmessage->err; -} - - -rd_kafka_t *test_create_handle(int mode, rd_kafka_conf_t *conf) { - rd_kafka_t *rk; - char errstr[512]; - - if (!conf) { - test_conf_init(&conf, NULL, 0); -#if WITH_SOCKEM - if (*test_sockem_conf) - test_socket_enable(conf); -#endif - } else { - if (!strcmp(test_conf_get(conf, "client.id"), "rdkafka")) - test_conf_set(conf, "client.id", test_curr->name); - } - - - - /* Creat kafka instance */ - rk = rd_kafka_new(mode, conf, errstr, sizeof(errstr)); - if (!rk) - TEST_FAIL("Failed to create rdkafka instance: %s\n", errstr); - - TEST_SAY("Created kafka instance %s\n", rd_kafka_name(rk)); - - return rk; -} - - -rd_kafka_t *test_create_producer(void) { - rd_kafka_conf_t *conf; - - test_conf_init(&conf, NULL, 0); - rd_kafka_conf_set_dr_msg_cb(conf, test_dr_msg_cb); - - return test_create_handle(RD_KAFKA_PRODUCER, conf); -} - - -/** - * Create topic_t object with va-arg list as key-value config pairs - * terminated by NULL. - */ -rd_kafka_topic_t * -test_create_topic_object(rd_kafka_t *rk, const char *topic, ...) { - rd_kafka_topic_t *rkt; - rd_kafka_topic_conf_t *topic_conf; - va_list ap; - const char *name, *val; - - test_conf_init(NULL, &topic_conf, 0); - - va_start(ap, topic); - while ((name = va_arg(ap, const char *)) && - (val = va_arg(ap, const char *))) { - test_topic_conf_set(topic_conf, name, val); - } - va_end(ap); - - rkt = rd_kafka_topic_new(rk, topic, topic_conf); - if (!rkt) - TEST_FAIL("Failed to create topic: %s\n", - rd_kafka_err2str(rd_kafka_last_error())); - - return rkt; -} - - -rd_kafka_topic_t * -test_create_producer_topic(rd_kafka_t *rk, const char *topic, ...) { - rd_kafka_topic_t *rkt; - rd_kafka_topic_conf_t *topic_conf; - char errstr[512]; - va_list ap; - const char *name, *val; - - test_conf_init(NULL, &topic_conf, 0); - - va_start(ap, topic); - while ((name = va_arg(ap, const char *)) && - (val = va_arg(ap, const char *))) { - if (rd_kafka_topic_conf_set(topic_conf, name, val, errstr, - sizeof(errstr)) != RD_KAFKA_CONF_OK) - TEST_FAIL("Conf failed: %s\n", errstr); - } - va_end(ap); - - /* Make sure all replicas are in-sync after producing - * so that consume test wont fail. */ - rd_kafka_topic_conf_set(topic_conf, "request.required.acks", "-1", - errstr, sizeof(errstr)); - - - rkt = rd_kafka_topic_new(rk, topic, topic_conf); - if (!rkt) - TEST_FAIL("Failed to create topic: %s\n", - rd_kafka_err2str(rd_kafka_last_error())); - - return rkt; -} - - - -/** - * Produces \p cnt messages and returns immediately. - * Does not wait for delivery. - * \p msgcounterp is incremented for each produced messages and passed - * as \p msg_opaque which is later used in test_dr_msg_cb to decrement - * the counter on delivery. - * - * If \p payload is NULL the message key and payload will be formatted - * according to standard test format, otherwise the key will be NULL and - * payload send as message payload. - * - * Default message size is 128 bytes, if \p size is non-zero and \p payload - * is NULL the message size of \p size will be used. - */ -void test_produce_msgs_nowait(rd_kafka_t *rk, - rd_kafka_topic_t *rkt, - uint64_t testid, - int32_t partition, - int msg_base, - int cnt, - const char *payload, - size_t size, - int msgrate, - int *msgcounterp) { - int msg_id; - test_timing_t t_all, t_poll; - char key[128]; - void *buf; - int64_t tot_bytes = 0; - int64_t tot_time_poll = 0; - int64_t per_msg_wait = 0; - - if (msgrate > 0) - per_msg_wait = 1000000 / (int64_t)msgrate; - - - if (payload) - buf = (void *)payload; - else { - if (size == 0) - size = 128; - buf = calloc(1, size); - } - - TEST_SAY("Produce to %s [%" PRId32 "]: messages #%d..%d\n", - rd_kafka_topic_name(rkt), partition, msg_base, msg_base + cnt); - - TIMING_START(&t_all, "PRODUCE"); - TIMING_START(&t_poll, "SUM(POLL)"); - - for (msg_id = msg_base; msg_id < msg_base + cnt; msg_id++) { - int wait_time = 0; - - if (!payload) - test_prepare_msg(testid, partition, msg_id, buf, size, - key, sizeof(key)); - - - if (rd_kafka_produce(rkt, partition, RD_KAFKA_MSG_F_COPY, buf, - size, !payload ? key : NULL, - !payload ? strlen(key) : 0, - msgcounterp) == -1) - TEST_FAIL( - "Failed to produce message %i " - "to partition %i: %s", - msg_id, (int)partition, - rd_kafka_err2str(rd_kafka_last_error())); - - (*msgcounterp)++; - tot_bytes += size; - - TIMING_RESTART(&t_poll); - do { - if (per_msg_wait) { - wait_time = (int)(per_msg_wait - - TIMING_DURATION(&t_poll)) / - 1000; - if (wait_time < 0) - wait_time = 0; - } - rd_kafka_poll(rk, wait_time); - } while (wait_time > 0); - - tot_time_poll = TIMING_DURATION(&t_poll); - - if (TIMING_EVERY(&t_all, 3 * 1000000)) - TEST_SAY( - "produced %3d%%: %d/%d messages " - "(%d msgs/s, %d bytes/s)\n", - ((msg_id - msg_base) * 100) / cnt, - msg_id - msg_base, cnt, - (int)((msg_id - msg_base) / - (TIMING_DURATION(&t_all) / 1000000)), - (int)((tot_bytes) / - (TIMING_DURATION(&t_all) / 1000000))); - } - - if (!payload) - free(buf); - - t_poll.duration = tot_time_poll; - TIMING_STOP(&t_poll); - TIMING_STOP(&t_all); -} - -/** - * Waits for the messages tracked by counter \p msgcounterp to be delivered. - */ -void test_wait_delivery(rd_kafka_t *rk, int *msgcounterp) { - test_timing_t t_all; - int start_cnt = *msgcounterp; - - TIMING_START(&t_all, "PRODUCE.DELIVERY.WAIT"); - - /* Wait for messages to be delivered */ - while (*msgcounterp > 0 && rd_kafka_outq_len(rk) > 0) { - rd_kafka_poll(rk, 10); - if (TIMING_EVERY(&t_all, 3 * 1000000)) { - int delivered = start_cnt - *msgcounterp; - TEST_SAY( - "wait_delivery: " - "%d/%d messages delivered: %d msgs/s\n", - delivered, start_cnt, - (int)(delivered / - (TIMING_DURATION(&t_all) / 1000000))); - } - } - - TIMING_STOP(&t_all); - - TEST_ASSERT(*msgcounterp == 0, - "Not all messages delivered: msgcounter still at %d, " - "outq_len %d", - *msgcounterp, rd_kafka_outq_len(rk)); -} - -/** - * Produces \p cnt messages and waits for succesful delivery - */ -void test_produce_msgs(rd_kafka_t *rk, - rd_kafka_topic_t *rkt, - uint64_t testid, - int32_t partition, - int msg_base, - int cnt, - const char *payload, - size_t size) { - int remains = 0; - - test_produce_msgs_nowait(rk, rkt, testid, partition, msg_base, cnt, - payload, size, 0, &remains); - - test_wait_delivery(rk, &remains); -} - - -/** - * @brief Produces \p cnt messages and waits for succesful delivery - */ -void test_produce_msgs2(rd_kafka_t *rk, - const char *topic, - uint64_t testid, - int32_t partition, - int msg_base, - int cnt, - const char *payload, - size_t size) { - int remains = 0; - rd_kafka_topic_t *rkt = test_create_topic_object(rk, topic, NULL); - - test_produce_msgs_nowait(rk, rkt, testid, partition, msg_base, cnt, - payload, size, 0, &remains); - - test_wait_delivery(rk, &remains); - - rd_kafka_topic_destroy(rkt); -} - -/** - * @brief Produces \p cnt messages without waiting for delivery. - */ -void test_produce_msgs2_nowait(rd_kafka_t *rk, - const char *topic, - uint64_t testid, - int32_t partition, - int msg_base, - int cnt, - const char *payload, - size_t size, - int *remainsp) { - rd_kafka_topic_t *rkt = test_create_topic_object(rk, topic, NULL); - - test_produce_msgs_nowait(rk, rkt, testid, partition, msg_base, cnt, - payload, size, 0, remainsp); - - rd_kafka_topic_destroy(rkt); -} - - -/** - * Produces \p cnt messages at \p msgs/s, and waits for succesful delivery - */ -void test_produce_msgs_rate(rd_kafka_t *rk, - rd_kafka_topic_t *rkt, - uint64_t testid, - int32_t partition, - int msg_base, - int cnt, - const char *payload, - size_t size, - int msgrate) { - int remains = 0; - - test_produce_msgs_nowait(rk, rkt, testid, partition, msg_base, cnt, - payload, size, msgrate, &remains); - - test_wait_delivery(rk, &remains); -} - - - -/** - * Create producer, produce \p msgcnt messages to \p topic \p partition, - * destroy consumer, and returns the used testid. - */ -uint64_t test_produce_msgs_easy_size(const char *topic, - uint64_t testid, - int32_t partition, - int msgcnt, - size_t size) { - rd_kafka_t *rk; - rd_kafka_topic_t *rkt; - test_timing_t t_produce; - - if (!testid) - testid = test_id_generate(); - rk = test_create_producer(); - rkt = test_create_producer_topic(rk, topic, NULL); - - TIMING_START(&t_produce, "PRODUCE"); - test_produce_msgs(rk, rkt, testid, partition, 0, msgcnt, NULL, size); - TIMING_STOP(&t_produce); - rd_kafka_topic_destroy(rkt); - rd_kafka_destroy(rk); - - return testid; -} - -rd_kafka_resp_err_t test_produce_sync(rd_kafka_t *rk, - rd_kafka_topic_t *rkt, - uint64_t testid, - int32_t partition) { - test_curr->produce_sync = 1; - test_produce_msgs(rk, rkt, testid, partition, 0, 1, NULL, 0); - test_curr->produce_sync = 0; - return test_curr->produce_sync_err; -} - - -/** - * @brief Easy produce function. - * - * @param ... is a NULL-terminated list of key, value config property pairs. - */ -void test_produce_msgs_easy_v(const char *topic, - uint64_t testid, - int32_t partition, - int msg_base, - int cnt, - size_t size, - ...) { - rd_kafka_conf_t *conf; - rd_kafka_t *p; - rd_kafka_topic_t *rkt; - va_list ap; - const char *key, *val; - - test_conf_init(&conf, NULL, 0); - - va_start(ap, size); - while ((key = va_arg(ap, const char *)) && - (val = va_arg(ap, const char *))) - test_conf_set(conf, key, val); - va_end(ap); - - rd_kafka_conf_set_dr_msg_cb(conf, test_dr_msg_cb); - - p = test_create_handle(RD_KAFKA_PRODUCER, conf); - - rkt = test_create_producer_topic(p, topic, NULL); - - test_produce_msgs(p, rkt, testid, partition, msg_base, cnt, NULL, size); - - rd_kafka_topic_destroy(rkt); - rd_kafka_destroy(p); -} - - -/** - * @brief Produce messages to multiple topic-partitions. - * - * @param ...vararg is a tuple of: - * const char *topic - * int32_t partition (or UA) - * int msg_base - * int msg_cnt - * - * End with a NULL topic - */ -void test_produce_msgs_easy_multi(uint64_t testid, ...) { - rd_kafka_conf_t *conf; - rd_kafka_t *p; - va_list ap; - const char *topic; - int msgcounter = 0; - - test_conf_init(&conf, NULL, 0); - - rd_kafka_conf_set_dr_msg_cb(conf, test_dr_msg_cb); - - p = test_create_handle(RD_KAFKA_PRODUCER, conf); - - va_start(ap, testid); - while ((topic = va_arg(ap, const char *))) { - int32_t partition = va_arg(ap, int32_t); - int msg_base = va_arg(ap, int); - int msg_cnt = va_arg(ap, int); - rd_kafka_topic_t *rkt; - - rkt = test_create_producer_topic(p, topic, NULL); - - test_produce_msgs_nowait(p, rkt, testid, partition, msg_base, - msg_cnt, NULL, 0, 0, &msgcounter); - - rd_kafka_topic_destroy(rkt); - } - va_end(ap); - - test_flush(p, tmout_multip(10 * 1000)); - - rd_kafka_destroy(p); -} - - - -/** - * @brief A standard incremental rebalance callback. - */ -void test_incremental_rebalance_cb(rd_kafka_t *rk, - rd_kafka_resp_err_t err, - rd_kafka_topic_partition_list_t *parts, - void *opaque) { - TEST_SAY("%s: incremental rebalance: %s: %d partition(s)%s\n", - rd_kafka_name(rk), rd_kafka_err2name(err), parts->cnt, - rd_kafka_assignment_lost(rk) ? ", assignment lost" : ""); - - switch (err) { - case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS: - test_consumer_incremental_assign("rebalance_cb", rk, parts); - break; - case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS: - test_consumer_incremental_unassign("rebalance_cb", rk, parts); - break; - default: - TEST_FAIL("Unknown rebalance event: %s", - rd_kafka_err2name(err)); - break; - } -} - -/** - * @brief A standard rebalance callback. - */ -void test_rebalance_cb(rd_kafka_t *rk, - rd_kafka_resp_err_t err, - rd_kafka_topic_partition_list_t *parts, - void *opaque) { - - if (!strcmp(rd_kafka_rebalance_protocol(rk), "COOPERATIVE")) { - test_incremental_rebalance_cb(rk, err, parts, opaque); - return; - } - - TEST_SAY("%s: Rebalance: %s: %d partition(s)\n", rd_kafka_name(rk), - rd_kafka_err2name(err), parts->cnt); - - switch (err) { - case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS: - test_consumer_assign("assign", rk, parts); - break; - case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS: - test_consumer_unassign("unassign", rk); - break; - default: - TEST_FAIL("Unknown rebalance event: %s", - rd_kafka_err2name(err)); - break; - } -} - - - -rd_kafka_t *test_create_consumer( - const char *group_id, - void (*rebalance_cb)(rd_kafka_t *rk, - rd_kafka_resp_err_t err, - rd_kafka_topic_partition_list_t *partitions, - void *opaque), - rd_kafka_conf_t *conf, - rd_kafka_topic_conf_t *default_topic_conf) { - rd_kafka_t *rk; - char tmp[64]; - - if (!conf) - test_conf_init(&conf, NULL, 0); - - if (group_id) { - test_conf_set(conf, "group.id", group_id); - - rd_snprintf(tmp, sizeof(tmp), "%d", test_session_timeout_ms); - test_conf_set(conf, "session.timeout.ms", tmp); - - if (rebalance_cb) - rd_kafka_conf_set_rebalance_cb(conf, rebalance_cb); - } else { - TEST_ASSERT(!rebalance_cb); - } - - if (default_topic_conf) - rd_kafka_conf_set_default_topic_conf(conf, default_topic_conf); - - /* Create kafka instance */ - rk = test_create_handle(RD_KAFKA_CONSUMER, conf); - - if (group_id) - rd_kafka_poll_set_consumer(rk); - - return rk; -} - -rd_kafka_topic_t *test_create_consumer_topic(rd_kafka_t *rk, - const char *topic) { - rd_kafka_topic_t *rkt; - rd_kafka_topic_conf_t *topic_conf; - - test_conf_init(NULL, &topic_conf, 0); - - rkt = rd_kafka_topic_new(rk, topic, topic_conf); - if (!rkt) - TEST_FAIL("Failed to create topic: %s\n", - rd_kafka_err2str(rd_kafka_last_error())); - - return rkt; -} - - -void test_consumer_start(const char *what, - rd_kafka_topic_t *rkt, - int32_t partition, - int64_t start_offset) { - - TEST_SAY("%s: consumer_start: %s [%" PRId32 "] at offset %" PRId64 "\n", - what, rd_kafka_topic_name(rkt), partition, start_offset); - - if (rd_kafka_consume_start(rkt, partition, start_offset) == -1) - TEST_FAIL("%s: consume_start failed: %s\n", what, - rd_kafka_err2str(rd_kafka_last_error())); -} - -void test_consumer_stop(const char *what, - rd_kafka_topic_t *rkt, - int32_t partition) { - - TEST_SAY("%s: consumer_stop: %s [%" PRId32 "]\n", what, - rd_kafka_topic_name(rkt), partition); - - if (rd_kafka_consume_stop(rkt, partition) == -1) - TEST_FAIL("%s: consume_stop failed: %s\n", what, - rd_kafka_err2str(rd_kafka_last_error())); -} - -void test_consumer_seek(const char *what, - rd_kafka_topic_t *rkt, - int32_t partition, - int64_t offset) { - int err; - - TEST_SAY("%s: consumer_seek: %s [%" PRId32 "] to offset %" PRId64 "\n", - what, rd_kafka_topic_name(rkt), partition, offset); - - if ((err = rd_kafka_seek(rkt, partition, offset, 2000))) - TEST_FAIL("%s: consume_seek(%s, %" PRId32 ", %" PRId64 - ") " - "failed: %s\n", - what, rd_kafka_topic_name(rkt), partition, offset, - rd_kafka_err2str(err)); -} - - - -/** - * Returns offset of the last message consumed - */ -int64_t test_consume_msgs(const char *what, - rd_kafka_topic_t *rkt, - uint64_t testid, - int32_t partition, - int64_t offset, - int exp_msg_base, - int exp_cnt, - int parse_fmt) { - int cnt = 0; - int msg_next = exp_msg_base; - int fails = 0; - int64_t offset_last = -1; - int64_t tot_bytes = 0; - test_timing_t t_first, t_all; - - TEST_SAY("%s: consume_msgs: %s [%" PRId32 - "]: expect msg #%d..%d " - "at offset %" PRId64 "\n", - what, rd_kafka_topic_name(rkt), partition, exp_msg_base, - exp_msg_base + exp_cnt, offset); - - if (offset != TEST_NO_SEEK) { - rd_kafka_resp_err_t err; - test_timing_t t_seek; - - TIMING_START(&t_seek, "SEEK"); - if ((err = rd_kafka_seek(rkt, partition, offset, 5000))) - TEST_FAIL("%s: consume_msgs: %s [%" PRId32 - "]: " - "seek to %" PRId64 " failed: %s\n", - what, rd_kafka_topic_name(rkt), partition, - offset, rd_kafka_err2str(err)); - TIMING_STOP(&t_seek); - TEST_SAY("%s: seeked to offset %" PRId64 "\n", what, offset); - } - - TIMING_START(&t_first, "FIRST MSG"); - TIMING_START(&t_all, "ALL MSGS"); - - while (cnt < exp_cnt) { - rd_kafka_message_t *rkmessage; - int msg_id; - - rkmessage = - rd_kafka_consume(rkt, partition, tmout_multip(5000)); - - if (TIMING_EVERY(&t_all, 3 * 1000000)) - TEST_SAY( - "%s: " - "consumed %3d%%: %d/%d messages " - "(%d msgs/s, %d bytes/s)\n", - what, cnt * 100 / exp_cnt, cnt, exp_cnt, - (int)(cnt / (TIMING_DURATION(&t_all) / 1000000)), - (int)(tot_bytes / - (TIMING_DURATION(&t_all) / 1000000))); - - if (!rkmessage) - TEST_FAIL("%s: consume_msgs: %s [%" PRId32 - "]: " - "expected msg #%d (%d/%d): timed out\n", - what, rd_kafka_topic_name(rkt), partition, - msg_next, cnt, exp_cnt); - - if (rkmessage->err) - TEST_FAIL("%s: consume_msgs: %s [%" PRId32 - "]: " - "expected msg #%d (%d/%d): got error: %s\n", - what, rd_kafka_topic_name(rkt), partition, - msg_next, cnt, exp_cnt, - rd_kafka_err2str(rkmessage->err)); - - if (cnt == 0) - TIMING_STOP(&t_first); - - if (parse_fmt) - test_msg_parse(testid, rkmessage, partition, &msg_id); - else - msg_id = 0; - - if (test_level >= 3) - TEST_SAY("%s: consume_msgs: %s [%" PRId32 - "]: " - "got msg #%d at offset %" PRId64 - " (expect #%d at offset %" PRId64 ")\n", - what, rd_kafka_topic_name(rkt), partition, - msg_id, rkmessage->offset, msg_next, - offset >= 0 ? offset + cnt : -1); - - if (parse_fmt && msg_id != msg_next) { - TEST_SAY("%s: consume_msgs: %s [%" PRId32 - "]: " - "expected msg #%d (%d/%d): got msg #%d\n", - what, rd_kafka_topic_name(rkt), partition, - msg_next, cnt, exp_cnt, msg_id); - fails++; - } - - cnt++; - tot_bytes += rkmessage->len; - msg_next++; - offset_last = rkmessage->offset; - - rd_kafka_message_destroy(rkmessage); - } - - TIMING_STOP(&t_all); - - if (fails) - TEST_FAIL("%s: consume_msgs: %s [%" PRId32 "]: %d failures\n", - what, rd_kafka_topic_name(rkt), partition, fails); - - TEST_SAY("%s: consume_msgs: %s [%" PRId32 - "]: " - "%d/%d messages consumed succesfully\n", - what, rd_kafka_topic_name(rkt), partition, cnt, exp_cnt); - return offset_last; -} - - -/** - * Create high-level consumer subscribing to \p topic from BEGINNING - * and expects \d exp_msgcnt with matching \p testid - * Destroys consumer when done. - * - * @param txn If true, isolation.level is set to read_committed. - * @param partition If -1 the topic will be subscribed to, otherwise the - * single partition will be assigned immediately. - * - * If \p group_id is NULL a new unique group is generated - */ -void test_consume_msgs_easy_mv0(const char *group_id, - const char *topic, - rd_bool_t txn, - int32_t partition, - uint64_t testid, - int exp_eofcnt, - int exp_msgcnt, - rd_kafka_topic_conf_t *tconf, - test_msgver_t *mv) { - rd_kafka_t *rk; - char grpid0[64]; - rd_kafka_conf_t *conf; - - test_conf_init(&conf, tconf ? NULL : &tconf, 0); - - if (!group_id) - group_id = test_str_id_generate(grpid0, sizeof(grpid0)); - - if (txn) - test_conf_set(conf, "isolation.level", "read_committed"); - - test_topic_conf_set(tconf, "auto.offset.reset", "smallest"); - if (exp_eofcnt != -1) - test_conf_set(conf, "enable.partition.eof", "true"); - rk = test_create_consumer(group_id, NULL, conf, tconf); - - rd_kafka_poll_set_consumer(rk); - - if (partition == -1) { - TEST_SAY( - "Subscribing to topic %s in group %s " - "(expecting %d msgs with testid %" PRIu64 ")\n", - topic, group_id, exp_msgcnt, testid); - - test_consumer_subscribe(rk, topic); - } else { - rd_kafka_topic_partition_list_t *plist; - - TEST_SAY("Assign topic %s [%" PRId32 - "] in group %s " - "(expecting %d msgs with testid %" PRIu64 ")\n", - topic, partition, group_id, exp_msgcnt, testid); - - plist = rd_kafka_topic_partition_list_new(1); - rd_kafka_topic_partition_list_add(plist, topic, partition); - test_consumer_assign("consume_easy_mv", rk, plist); - rd_kafka_topic_partition_list_destroy(plist); - } - - /* Consume messages */ - test_consumer_poll("consume.easy", rk, testid, exp_eofcnt, -1, - exp_msgcnt, mv); - - test_consumer_close(rk); - - rd_kafka_destroy(rk); -} - -void test_consume_msgs_easy(const char *group_id, - const char *topic, - uint64_t testid, - int exp_eofcnt, - int exp_msgcnt, - rd_kafka_topic_conf_t *tconf) { - test_msgver_t mv; - - test_msgver_init(&mv, testid); - - test_consume_msgs_easy_mv(group_id, topic, -1, testid, exp_eofcnt, - exp_msgcnt, tconf, &mv); - - test_msgver_clear(&mv); -} - - -void test_consume_txn_msgs_easy(const char *group_id, - const char *topic, - uint64_t testid, - int exp_eofcnt, - int exp_msgcnt, - rd_kafka_topic_conf_t *tconf) { - test_msgver_t mv; - - test_msgver_init(&mv, testid); - - test_consume_msgs_easy_mv0(group_id, topic, rd_true /*txn*/, -1, testid, - exp_eofcnt, exp_msgcnt, tconf, &mv); - - test_msgver_clear(&mv); -} - - -/** - * @brief Waits for up to \p timeout_ms for consumer to receive assignment. - * If no assignment received without the timeout the test fails. - * - * @warning This method will poll the consumer and might thus read messages. - * Set \p do_poll to false to use a sleep rather than poll. - */ -void test_consumer_wait_assignment(rd_kafka_t *rk, rd_bool_t do_poll) { - rd_kafka_topic_partition_list_t *assignment = NULL; - int i; - - while (1) { - rd_kafka_resp_err_t err; - - err = rd_kafka_assignment(rk, &assignment); - TEST_ASSERT(!err, "rd_kafka_assignment() failed: %s", - rd_kafka_err2str(err)); - - if (assignment->cnt > 0) - break; - - rd_kafka_topic_partition_list_destroy(assignment); - - if (do_poll) - test_consumer_poll_once(rk, NULL, 1000); - else - rd_usleep(1000 * 1000, NULL); - } - - TEST_SAY("%s: Assignment (%d partition(s)): ", rd_kafka_name(rk), - assignment->cnt); - for (i = 0; i < assignment->cnt; i++) - TEST_SAY0("%s%s[%" PRId32 "]", i == 0 ? "" : ", ", - assignment->elems[i].topic, - assignment->elems[i].partition); - TEST_SAY0("\n"); - - rd_kafka_topic_partition_list_destroy(assignment); -} - - -/** - * @brief Verify that the consumer's assignment matches the expected assignment. - * - * The va-list is a NULL-terminated list of (const char *topic, int partition) - * tuples. - * - * Fails the test on mismatch, unless \p fail_immediately is false. - */ -void test_consumer_verify_assignment0(const char *func, - int line, - rd_kafka_t *rk, - int fail_immediately, - ...) { - va_list ap; - int cnt = 0; - const char *topic; - rd_kafka_topic_partition_list_t *assignment; - rd_kafka_resp_err_t err; - int i; - - if ((err = rd_kafka_assignment(rk, &assignment))) - TEST_FAIL("%s:%d: Failed to get assignment for %s: %s", func, - line, rd_kafka_name(rk), rd_kafka_err2str(err)); - - TEST_SAY("%s assignment (%d partition(s)):\n", rd_kafka_name(rk), - assignment->cnt); - for (i = 0; i < assignment->cnt; i++) - TEST_SAY(" %s [%" PRId32 "]\n", assignment->elems[i].topic, - assignment->elems[i].partition); - - va_start(ap, fail_immediately); - while ((topic = va_arg(ap, const char *))) { - int partition = va_arg(ap, int); - cnt++; - - if (!rd_kafka_topic_partition_list_find(assignment, topic, - partition)) - TEST_FAIL_LATER( - "%s:%d: Expected %s [%d] not found in %s's " - "assignment (%d partition(s))", - func, line, topic, partition, rd_kafka_name(rk), - assignment->cnt); - } - va_end(ap); - - if (cnt != assignment->cnt) - TEST_FAIL_LATER( - "%s:%d: " - "Expected %d assigned partition(s) for %s, not %d", - func, line, cnt, rd_kafka_name(rk), assignment->cnt); - - if (fail_immediately) - TEST_LATER_CHECK(); - - rd_kafka_topic_partition_list_destroy(assignment); -} - - - -/** - * @brief Start subscribing for 'topic' - */ -void test_consumer_subscribe(rd_kafka_t *rk, const char *topic) { - rd_kafka_topic_partition_list_t *topics; - rd_kafka_resp_err_t err; - - topics = rd_kafka_topic_partition_list_new(1); - rd_kafka_topic_partition_list_add(topics, topic, RD_KAFKA_PARTITION_UA); - - err = rd_kafka_subscribe(rk, topics); - if (err) - TEST_FAIL("%s: Failed to subscribe to %s: %s\n", - rd_kafka_name(rk), topic, rd_kafka_err2str(err)); - - rd_kafka_topic_partition_list_destroy(topics); -} - - -void test_consumer_assign(const char *what, - rd_kafka_t *rk, - rd_kafka_topic_partition_list_t *partitions) { - rd_kafka_resp_err_t err; - test_timing_t timing; - - TIMING_START(&timing, "ASSIGN.PARTITIONS"); - err = rd_kafka_assign(rk, partitions); - TIMING_STOP(&timing); - if (err) - TEST_FAIL("%s: failed to assign %d partition(s): %s\n", what, - partitions->cnt, rd_kafka_err2str(err)); - else - TEST_SAY("%s: assigned %d partition(s)\n", what, - partitions->cnt); -} - - -void test_consumer_incremental_assign( - const char *what, - rd_kafka_t *rk, - rd_kafka_topic_partition_list_t *partitions) { - rd_kafka_error_t *error; - test_timing_t timing; - - TIMING_START(&timing, "INCREMENTAL.ASSIGN.PARTITIONS"); - error = rd_kafka_incremental_assign(rk, partitions); - TIMING_STOP(&timing); - if (error) { - TEST_FAIL( - "%s: incremental assign of %d partition(s) failed: " - "%s", - what, partitions->cnt, rd_kafka_error_string(error)); - rd_kafka_error_destroy(error); - } else - TEST_SAY("%s: incremental assign of %d partition(s) done\n", - what, partitions->cnt); -} - - -void test_consumer_unassign(const char *what, rd_kafka_t *rk) { - rd_kafka_resp_err_t err; - test_timing_t timing; - - TIMING_START(&timing, "UNASSIGN.PARTITIONS"); - err = rd_kafka_assign(rk, NULL); - TIMING_STOP(&timing); - if (err) - TEST_FAIL("%s: failed to unassign current partitions: %s\n", - what, rd_kafka_err2str(err)); - else - TEST_SAY("%s: unassigned current partitions\n", what); -} - - -void test_consumer_incremental_unassign( - const char *what, - rd_kafka_t *rk, - rd_kafka_topic_partition_list_t *partitions) { - rd_kafka_error_t *error; - test_timing_t timing; - - TIMING_START(&timing, "INCREMENTAL.UNASSIGN.PARTITIONS"); - error = rd_kafka_incremental_unassign(rk, partitions); - TIMING_STOP(&timing); - if (error) { - TEST_FAIL( - "%s: incremental unassign of %d partition(s) " - "failed: %s", - what, partitions->cnt, rd_kafka_error_string(error)); - rd_kafka_error_destroy(error); - } else - TEST_SAY("%s: incremental unassign of %d partition(s) done\n", - what, partitions->cnt); -} - - -/** - * @brief Assign a single partition with an optional starting offset - */ -void test_consumer_assign_partition(const char *what, - rd_kafka_t *rk, - const char *topic, - int32_t partition, - int64_t offset) { - rd_kafka_topic_partition_list_t *part; - - part = rd_kafka_topic_partition_list_new(1); - rd_kafka_topic_partition_list_add(part, topic, partition)->offset = - offset; - - test_consumer_assign(what, rk, part); - - rd_kafka_topic_partition_list_destroy(part); -} - - -void test_consumer_pause_resume_partition(rd_kafka_t *rk, - const char *topic, - int32_t partition, - rd_bool_t pause) { - rd_kafka_topic_partition_list_t *part; - rd_kafka_resp_err_t err; - - part = rd_kafka_topic_partition_list_new(1); - rd_kafka_topic_partition_list_add(part, topic, partition); - - if (pause) - err = rd_kafka_pause_partitions(rk, part); - else - err = rd_kafka_resume_partitions(rk, part); - - TEST_ASSERT(!err, "Failed to %s %s [%" PRId32 "]: %s", - pause ? "pause" : "resume", topic, partition, - rd_kafka_err2str(err)); - - rd_kafka_topic_partition_list_destroy(part); -} - - -/** - * Message verification services - * - */ - -void test_msgver_init(test_msgver_t *mv, uint64_t testid) { - memset(mv, 0, sizeof(*mv)); - mv->testid = testid; - /* Max warning logs before suppressing. */ - mv->log_max = (test_level + 1) * 100; -} - -void test_msgver_ignore_eof(test_msgver_t *mv) { - mv->ignore_eof = rd_true; -} - -#define TEST_MV_WARN(mv, ...) \ - do { \ - if ((mv)->log_cnt++ > (mv)->log_max) \ - (mv)->log_suppr_cnt++; \ - else \ - TEST_WARN(__VA_ARGS__); \ - } while (0) - - - -static void test_mv_mvec_grow(struct test_mv_mvec *mvec, int tot_size) { - if (tot_size <= mvec->size) - return; - mvec->size = tot_size; - mvec->m = realloc(mvec->m, sizeof(*mvec->m) * mvec->size); -} - -/** - * Make sure there is room for at least \p cnt messages, else grow mvec. - */ -static void test_mv_mvec_reserve(struct test_mv_mvec *mvec, int cnt) { - test_mv_mvec_grow(mvec, mvec->cnt + cnt); -} - -void test_mv_mvec_init(struct test_mv_mvec *mvec, int exp_cnt) { - TEST_ASSERT(mvec->m == NULL, "mvec not cleared"); - - if (!exp_cnt) - return; - - test_mv_mvec_grow(mvec, exp_cnt); -} - - -void test_mv_mvec_clear(struct test_mv_mvec *mvec) { - if (mvec->m) - free(mvec->m); -} - -void test_msgver_clear(test_msgver_t *mv) { - int i; - for (i = 0; i < mv->p_cnt; i++) { - struct test_mv_p *p = mv->p[i]; - free(p->topic); - test_mv_mvec_clear(&p->mvec); - free(p); - } - - free(mv->p); - - test_msgver_init(mv, mv->testid); -} - -struct test_mv_p *test_msgver_p_get(test_msgver_t *mv, - const char *topic, - int32_t partition, - int do_create) { - int i; - struct test_mv_p *p; - - for (i = 0; i < mv->p_cnt; i++) { - p = mv->p[i]; - if (p->partition == partition && !strcmp(p->topic, topic)) - return p; - } - - if (!do_create) - TEST_FAIL("Topic %s [%d] not found in msgver", topic, - partition); - - if (mv->p_cnt == mv->p_size) { - mv->p_size = (mv->p_size + 4) * 2; - mv->p = realloc(mv->p, sizeof(*mv->p) * mv->p_size); - } - - mv->p[mv->p_cnt++] = p = calloc(1, sizeof(*p)); - - p->topic = rd_strdup(topic); - p->partition = partition; - p->eof_offset = RD_KAFKA_OFFSET_INVALID; - - return p; -} - - -/** - * Add (room for) message to message vector. - * Resizes the vector as needed. - */ -static struct test_mv_m *test_mv_mvec_add(struct test_mv_mvec *mvec) { - if (mvec->cnt == mvec->size) { - test_mv_mvec_grow(mvec, (mvec->size ? mvec->size * 2 : 10000)); - } - - mvec->cnt++; - - return &mvec->m[mvec->cnt - 1]; -} - -/** - * Returns message at index \p mi - */ -static RD_INLINE struct test_mv_m *test_mv_mvec_get(struct test_mv_mvec *mvec, - int mi) { - if (mi >= mvec->cnt) - return NULL; - return &mvec->m[mi]; -} - -/** - * @returns the message with msgid \p msgid, or NULL. - */ -static struct test_mv_m *test_mv_mvec_find_by_msgid(struct test_mv_mvec *mvec, - int msgid) { - int mi; - - for (mi = 0; mi < mvec->cnt; mi++) - if (mvec->m[mi].msgid == msgid) - return &mvec->m[mi]; - - return NULL; -} - - -/** - * Print message list to \p fp - */ -static RD_UNUSED void test_mv_mvec_dump(FILE *fp, - const struct test_mv_mvec *mvec) { - int mi; - - fprintf(fp, "*** Dump mvec with %d messages (capacity %d): ***\n", - mvec->cnt, mvec->size); - for (mi = 0; mi < mvec->cnt; mi++) - fprintf(fp, " msgid %d, offset %" PRId64 "\n", - mvec->m[mi].msgid, mvec->m[mi].offset); - fprintf(fp, "*** Done ***\n"); -} - -static void test_mv_mvec_sort(struct test_mv_mvec *mvec, - int (*cmp)(const void *, const void *)) { - qsort(mvec->m, mvec->cnt, sizeof(*mvec->m), cmp); -} - - -/** - * @brief Adds a message to the msgver service. - * - * @returns 1 if message is from the expected testid, else 0 (not added) - */ -int test_msgver_add_msg00(const char *func, - int line, - const char *clientname, - test_msgver_t *mv, - uint64_t testid, - const char *topic, - int32_t partition, - int64_t offset, - int64_t timestamp, - int32_t broker_id, - rd_kafka_resp_err_t err, - int msgnum) { - struct test_mv_p *p; - struct test_mv_m *m; - - if (testid != mv->testid) { - TEST_SAYL(3, - "%s:%d: %s: mismatching testid %" PRIu64 - " != %" PRIu64 "\n", - func, line, clientname, testid, mv->testid); - return 0; /* Ignore message */ - } - - if (err == RD_KAFKA_RESP_ERR__PARTITION_EOF && mv->ignore_eof) { - TEST_SAYL(3, "%s:%d: %s: ignoring EOF for %s [%" PRId32 "]\n", - func, line, clientname, topic, partition); - return 0; /* Ignore message */ - } - - p = test_msgver_p_get(mv, topic, partition, 1); - - if (err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { - p->eof_offset = offset; - return 1; - } - - m = test_mv_mvec_add(&p->mvec); - - m->offset = offset; - m->msgid = msgnum; - m->timestamp = timestamp; - m->broker_id = broker_id; - - if (test_level > 2) { - TEST_SAY( - "%s:%d: %s: " - "Recv msg %s [%" PRId32 "] offset %" PRId64 - " msgid %d " - "timestamp %" PRId64 " broker %" PRId32 "\n", - func, line, clientname, p->topic, p->partition, m->offset, - m->msgid, m->timestamp, m->broker_id); - } - - mv->msgcnt++; - - return 1; -} - -/** - * Adds a message to the msgver service. - * - * Message must be a proper message or PARTITION_EOF. - * - * @param override_topic if non-NULL, overrides the rkmessage's topic - * with this one. - * - * @returns 1 if message is from the expected testid, else 0 (not added). - */ -int test_msgver_add_msg0(const char *func, - int line, - const char *clientname, - test_msgver_t *mv, - const rd_kafka_message_t *rkmessage, - const char *override_topic) { - uint64_t in_testid; - int in_part; - int in_msgnum = -1; - char buf[128]; - const void *val; - size_t valsize; - - if (mv->fwd) - test_msgver_add_msg0(func, line, clientname, mv->fwd, rkmessage, - override_topic); - - if (rd_kafka_message_status(rkmessage) == - RD_KAFKA_MSG_STATUS_NOT_PERSISTED && - rkmessage->err) { - if (rkmessage->err != RD_KAFKA_RESP_ERR__PARTITION_EOF) - return 0; /* Ignore error */ - - in_testid = mv->testid; - - } else { - - if (!mv->msgid_hdr) { - rd_snprintf(buf, sizeof(buf), "%.*s", - (int)rkmessage->len, - (char *)rkmessage->payload); - val = buf; - } else { - /* msgid is in message header */ - rd_kafka_headers_t *hdrs; - - if (rd_kafka_message_headers(rkmessage, &hdrs) || - rd_kafka_header_get_last(hdrs, mv->msgid_hdr, &val, - &valsize)) { - TEST_SAYL(3, - "%s:%d: msgid expected in header %s " - "but %s exists for " - "message at offset %" PRId64 - " has no headers\n", - func, line, mv->msgid_hdr, - hdrs ? "no such header" - : "no headers", - rkmessage->offset); - - return 0; - } - } - - if (sscanf(val, "testid=%" SCNu64 ", partition=%i, msg=%i\n", - &in_testid, &in_part, &in_msgnum) != 3) - TEST_FAIL( - "%s:%d: Incorrect format at offset %" PRId64 ": %s", - func, line, rkmessage->offset, (const char *)val); - } - - return test_msgver_add_msg00( - func, line, clientname, mv, in_testid, - override_topic ? override_topic - : rd_kafka_topic_name(rkmessage->rkt), - rkmessage->partition, rkmessage->offset, - rd_kafka_message_timestamp(rkmessage, NULL), - rd_kafka_message_broker_id(rkmessage), rkmessage->err, in_msgnum); - return 1; -} - - - -/** - * Verify that all messages were received in order. - * - * - Offsets need to occur without gaps - * - msgids need to be increasing: but may have gaps, e.g., using partitioner) - */ -static int test_mv_mvec_verify_order(test_msgver_t *mv, - int flags, - struct test_mv_p *p, - struct test_mv_mvec *mvec, - struct test_mv_vs *vs) { - int mi; - int fails = 0; - - for (mi = 1 /*skip first*/; mi < mvec->cnt; mi++) { - struct test_mv_m *prev = test_mv_mvec_get(mvec, mi - 1); - struct test_mv_m *this = test_mv_mvec_get(mvec, mi); - - if (((flags & TEST_MSGVER_BY_OFFSET) && - prev->offset + 1 != this->offset) || - ((flags & TEST_MSGVER_BY_MSGID) && - prev->msgid > this->msgid)) { - TEST_MV_WARN(mv, - " %s [%" PRId32 - "] msg rcvidx #%d/%d: " - "out of order (prev vs this): " - "offset %" PRId64 " vs %" PRId64 - ", " - "msgid %d vs %d\n", - p ? p->topic : "*", p ? p->partition : -1, - mi, mvec->cnt, prev->offset, this->offset, - prev->msgid, this->msgid); - fails++; - } else if ((flags & TEST_MSGVER_BY_BROKER_ID) && - this->broker_id != vs->broker_id) { - TEST_MV_WARN(mv, - " %s [%" PRId32 - "] msg rcvidx #%d/%d: " - "broker id mismatch: expected %" PRId32 - ", not %" PRId32 "\n", - p ? p->topic : "*", p ? p->partition : -1, - mi, mvec->cnt, vs->broker_id, - this->broker_id); - fails++; - } - } - - return fails; -} - - -/** - * @brief Verify that messages correspond to 'correct' msgver. - */ -static int test_mv_mvec_verify_corr(test_msgver_t *mv, - int flags, - struct test_mv_p *p, - struct test_mv_mvec *mvec, - struct test_mv_vs *vs) { - int mi; - int fails = 0; - struct test_mv_p *corr_p = NULL; - struct test_mv_mvec *corr_mvec; - int verifycnt = 0; - - TEST_ASSERT(vs->corr); - - /* Get correct mvec for comparison. */ - if (p) - corr_p = test_msgver_p_get(vs->corr, p->topic, p->partition, 0); - if (!corr_p) { - TEST_MV_WARN(mv, - " %s [%" PRId32 - "]: " - "no corresponding correct partition found\n", - p ? p->topic : "*", p ? p->partition : -1); - return 1; - } - - corr_mvec = &corr_p->mvec; - - for (mi = 0; mi < mvec->cnt; mi++) { - struct test_mv_m *this = test_mv_mvec_get(mvec, mi); - const struct test_mv_m *corr; - - - if (flags & TEST_MSGVER_SUBSET) - corr = - test_mv_mvec_find_by_msgid(corr_mvec, this->msgid); - else - corr = test_mv_mvec_get(corr_mvec, mi); - - if (0) - TEST_MV_WARN(mv, - "msg #%d: msgid %d, offset %" PRId64 "\n", - mi, this->msgid, this->offset); - if (!corr) { - if (!(flags & TEST_MSGVER_SUBSET)) { - TEST_MV_WARN( - mv, - " %s [%" PRId32 - "] msg rcvidx #%d/%d: " - "out of range: correct mvec has " - "%d messages: " - "message offset %" PRId64 ", msgid %d\n", - p ? p->topic : "*", p ? p->partition : -1, - mi, mvec->cnt, corr_mvec->cnt, this->offset, - this->msgid); - fails++; - } - continue; - } - - if (((flags & TEST_MSGVER_BY_OFFSET) && - this->offset != corr->offset) || - ((flags & TEST_MSGVER_BY_MSGID) && - this->msgid != corr->msgid) || - ((flags & TEST_MSGVER_BY_TIMESTAMP) && - this->timestamp != corr->timestamp) || - ((flags & TEST_MSGVER_BY_BROKER_ID) && - this->broker_id != corr->broker_id)) { - TEST_MV_WARN( - mv, - " %s [%" PRId32 - "] msg rcvidx #%d/%d: " - "did not match correct msg: " - "offset %" PRId64 " vs %" PRId64 - ", " - "msgid %d vs %d, " - "timestamp %" PRId64 " vs %" PRId64 - ", " - "broker %" PRId32 " vs %" PRId32 " (fl 0x%x)\n", - p ? p->topic : "*", p ? p->partition : -1, mi, - mvec->cnt, this->offset, corr->offset, this->msgid, - corr->msgid, this->timestamp, corr->timestamp, - this->broker_id, corr->broker_id, flags); - fails++; - } else { - verifycnt++; - } - } - - if (verifycnt != corr_mvec->cnt && !(flags & TEST_MSGVER_SUBSET)) { - TEST_MV_WARN(mv, - " %s [%" PRId32 - "]: of %d input messages, " - "only %d/%d matched correct messages\n", - p ? p->topic : "*", p ? p->partition : -1, - mvec->cnt, verifycnt, corr_mvec->cnt); - fails++; - } - - return fails; -} - - - -static int test_mv_m_cmp_offset(const void *_a, const void *_b) { - const struct test_mv_m *a = _a, *b = _b; - - return RD_CMP(a->offset, b->offset); -} - -static int test_mv_m_cmp_msgid(const void *_a, const void *_b) { - const struct test_mv_m *a = _a, *b = _b; - - return RD_CMP(a->msgid, b->msgid); -} - - -/** - * Verify that there are no duplicate message. - * - * - Offsets are checked - * - msgids are checked - * - * * NOTE: This sorts the message (.m) array, first by offset, then by msgid - * and leaves the message array sorted (by msgid) - */ -static int test_mv_mvec_verify_dup(test_msgver_t *mv, - int flags, - struct test_mv_p *p, - struct test_mv_mvec *mvec, - struct test_mv_vs *vs) { - int mi; - int fails = 0; - enum { _P_OFFSET, _P_MSGID } pass; - - for (pass = _P_OFFSET; pass <= _P_MSGID; pass++) { - - if (pass == _P_OFFSET) { - if (!(flags & TEST_MSGVER_BY_OFFSET)) - continue; - test_mv_mvec_sort(mvec, test_mv_m_cmp_offset); - } else if (pass == _P_MSGID) { - if (!(flags & TEST_MSGVER_BY_MSGID)) - continue; - test_mv_mvec_sort(mvec, test_mv_m_cmp_msgid); - } - - for (mi = 1 /*skip first*/; mi < mvec->cnt; mi++) { - struct test_mv_m *prev = test_mv_mvec_get(mvec, mi - 1); - struct test_mv_m *this = test_mv_mvec_get(mvec, mi); - int is_dup = 0; - - if (pass == _P_OFFSET) - is_dup = prev->offset == this->offset; - else if (pass == _P_MSGID) - is_dup = prev->msgid == this->msgid; - - if (!is_dup) - continue; - - TEST_MV_WARN(mv, - " %s [%" PRId32 - "] " - "duplicate msg (prev vs this): " - "offset %" PRId64 " vs %" PRId64 - ", " - "msgid %d vs %d\n", - p ? p->topic : "*", p ? p->partition : -1, - prev->offset, this->offset, prev->msgid, - this->msgid); - fails++; - } - } - - return fails; -} - -/** - * @brief Verify that all messages are from the correct broker. - */ -static int test_mv_mvec_verify_broker(test_msgver_t *mv, - int flags, - struct test_mv_p *p, - struct test_mv_mvec *mvec, - struct test_mv_vs *vs) { - int mi; - int fails = 0; - - /* Assume that the correct flag has been checked already. */ - - - rd_assert(flags & TEST_MSGVER_BY_BROKER_ID); - for (mi = 0; mi < mvec->cnt; mi++) { - struct test_mv_m *this = test_mv_mvec_get(mvec, mi); - if (this->broker_id != vs->broker_id) { - TEST_MV_WARN( - mv, - " %s [%" PRId32 - "] broker_id check: " - "msgid #%d (at mi %d): " - "broker_id %" PRId32 - " is not the expected broker_id %" PRId32 "\n", - p ? p->topic : "*", p ? p->partition : -1, - this->msgid, mi, this->broker_id, vs->broker_id); - fails++; - } - } - return fails; -} - - -/** - * Verify that \p mvec contains the expected range: - * - TEST_MSGVER_BY_MSGID: msgid within \p vs->msgid_min .. \p vs->msgid_max - * - TEST_MSGVER_BY_TIMESTAMP: timestamp with \p vs->timestamp_min .. _max - * - * * NOTE: TEST_MSGVER_BY_MSGID is required - * - * * NOTE: This sorts the message (.m) array by msgid - * and leaves the message array sorted (by msgid) - */ -static int test_mv_mvec_verify_range(test_msgver_t *mv, - int flags, - struct test_mv_p *p, - struct test_mv_mvec *mvec, - struct test_mv_vs *vs) { - int mi; - int fails = 0; - int cnt = 0; - int exp_cnt = vs->msgid_max - vs->msgid_min + 1; - int skip_cnt = 0; - - if (!(flags & TEST_MSGVER_BY_MSGID)) - return 0; - - test_mv_mvec_sort(mvec, test_mv_m_cmp_msgid); - - // test_mv_mvec_dump(stdout, mvec); - - for (mi = 0; mi < mvec->cnt; mi++) { - struct test_mv_m *prev = - mi ? test_mv_mvec_get(mvec, mi - 1) : NULL; - struct test_mv_m *this = test_mv_mvec_get(mvec, mi); - - if (this->msgid < vs->msgid_min) { - skip_cnt++; - continue; - } else if (this->msgid > vs->msgid_max) - break; - - if (flags & TEST_MSGVER_BY_TIMESTAMP) { - if (this->timestamp < vs->timestamp_min || - this->timestamp > vs->timestamp_max) { - TEST_MV_WARN( - mv, - " %s [%" PRId32 - "] range check: " - "msgid #%d (at mi %d): " - "timestamp %" PRId64 - " outside " - "expected range %" PRId64 "..%" PRId64 "\n", - p ? p->topic : "*", p ? p->partition : -1, - this->msgid, mi, this->timestamp, - vs->timestamp_min, vs->timestamp_max); - fails++; - } - } - - if ((flags & TEST_MSGVER_BY_BROKER_ID) && - this->broker_id != vs->broker_id) { - TEST_MV_WARN( - mv, - " %s [%" PRId32 - "] range check: " - "msgid #%d (at mi %d): " - "expected broker id %" PRId32 ", not %" PRId32 "\n", - p ? p->topic : "*", p ? p->partition : -1, - this->msgid, mi, vs->broker_id, this->broker_id); - fails++; - } - - if (cnt++ == 0) { - if (this->msgid != vs->msgid_min) { - TEST_MV_WARN(mv, - " %s [%" PRId32 - "] range check: " - "first message #%d (at mi %d) " - "is not first in " - "expected range %d..%d\n", - p ? p->topic : "*", - p ? p->partition : -1, this->msgid, - mi, vs->msgid_min, vs->msgid_max); - fails++; - } - } else if (cnt > exp_cnt) { - TEST_MV_WARN(mv, - " %s [%" PRId32 - "] range check: " - "too many messages received (%d/%d) at " - "msgid %d for expected range %d..%d\n", - p ? p->topic : "*", p ? p->partition : -1, - cnt, exp_cnt, this->msgid, vs->msgid_min, - vs->msgid_max); - fails++; - } - - if (!prev) { - skip_cnt++; - continue; - } - - if (prev->msgid + 1 != this->msgid) { - TEST_MV_WARN(mv, - " %s [%" PRId32 - "] range check: " - " %d message(s) missing between " - "msgid %d..%d in expected range %d..%d\n", - p ? p->topic : "*", p ? p->partition : -1, - this->msgid - prev->msgid - 1, - prev->msgid + 1, this->msgid - 1, - vs->msgid_min, vs->msgid_max); - fails++; - } - } - - if (cnt != exp_cnt) { - TEST_MV_WARN(mv, - " %s [%" PRId32 - "] range check: " - " wrong number of messages seen, wanted %d got %d " - "in expected range %d..%d (%d messages skipped)\n", - p ? p->topic : "*", p ? p->partition : -1, exp_cnt, - cnt, vs->msgid_min, vs->msgid_max, skip_cnt); - fails++; - } - - return fails; -} - - - -/** - * Run verifier \p f for all partitions. - */ -#define test_mv_p_verify_f(mv, flags, f, vs) \ - test_mv_p_verify_f0(mv, flags, f, #f, vs) -static int test_mv_p_verify_f0(test_msgver_t *mv, - int flags, - int (*f)(test_msgver_t *mv, - int flags, - struct test_mv_p *p, - struct test_mv_mvec *mvec, - struct test_mv_vs *vs), - const char *f_name, - struct test_mv_vs *vs) { - int i; - int fails = 0; - - for (i = 0; i < mv->p_cnt; i++) { - TEST_SAY("Verifying %s [%" PRId32 "] %d msgs with %s\n", - mv->p[i]->topic, mv->p[i]->partition, - mv->p[i]->mvec.cnt, f_name); - fails += f(mv, flags, mv->p[i], &mv->p[i]->mvec, vs); - } - - return fails; -} - - -/** - * Collect all messages from all topics and partitions into vs->mvec - */ -static void test_mv_collect_all_msgs(test_msgver_t *mv, struct test_mv_vs *vs) { - int i; - - for (i = 0; i < mv->p_cnt; i++) { - struct test_mv_p *p = mv->p[i]; - int mi; - - test_mv_mvec_reserve(&vs->mvec, p->mvec.cnt); - for (mi = 0; mi < p->mvec.cnt; mi++) { - struct test_mv_m *m = test_mv_mvec_get(&p->mvec, mi); - struct test_mv_m *m_new = test_mv_mvec_add(&vs->mvec); - *m_new = *m; - } - } -} - - -/** - * Verify that all messages (by msgid) in range msg_base+exp_cnt were received - * and received only once. - * This works across all partitions. - */ -static int -test_msgver_verify_range(test_msgver_t *mv, int flags, struct test_mv_vs *vs) { - int fails = 0; - - /** - * Create temporary array to hold expected message set, - * then traverse all topics and partitions and move matching messages - * to that set. Then verify the message set. - */ - - test_mv_mvec_init(&vs->mvec, vs->exp_cnt); - - /* Collect all msgs into vs mvec */ - test_mv_collect_all_msgs(mv, vs); - - fails += test_mv_mvec_verify_range(mv, TEST_MSGVER_BY_MSGID | flags, - NULL, &vs->mvec, vs); - fails += test_mv_mvec_verify_dup(mv, TEST_MSGVER_BY_MSGID | flags, NULL, - &vs->mvec, vs); - - test_mv_mvec_clear(&vs->mvec); - - return fails; -} - - -/** - * Verify that \p exp_cnt messages were received for \p topic and \p partition - * starting at msgid base \p msg_base. - */ -int test_msgver_verify_part0(const char *func, - int line, - const char *what, - test_msgver_t *mv, - int flags, - const char *topic, - int partition, - int msg_base, - int exp_cnt) { - int fails = 0; - struct test_mv_vs vs = {.msg_base = msg_base, .exp_cnt = exp_cnt}; - struct test_mv_p *p; - - TEST_SAY( - "%s:%d: %s: Verifying %d received messages (flags 0x%x) " - "in %s [%d]: expecting msgids %d..%d (%d)\n", - func, line, what, mv->msgcnt, flags, topic, partition, msg_base, - msg_base + exp_cnt, exp_cnt); - - p = test_msgver_p_get(mv, topic, partition, 0); - - /* Per-partition checks */ - if (flags & TEST_MSGVER_ORDER) - fails += test_mv_mvec_verify_order(mv, flags, p, &p->mvec, &vs); - if (flags & TEST_MSGVER_DUP) - fails += test_mv_mvec_verify_dup(mv, flags, p, &p->mvec, &vs); - - if (mv->msgcnt < vs.exp_cnt) { - TEST_MV_WARN(mv, - "%s:%d: " - "%s [%" PRId32 - "] expected %d messages but only " - "%d received\n", - func, line, p ? p->topic : "*", - p ? p->partition : -1, vs.exp_cnt, mv->msgcnt); - fails++; - } - - - if (mv->log_suppr_cnt > 0) - TEST_WARN("%s:%d: %s: %d message warning logs suppressed\n", - func, line, what, mv->log_suppr_cnt); - - if (fails) - TEST_FAIL( - "%s:%d: %s: Verification of %d received messages " - "failed: " - "expected msgids %d..%d (%d): see previous errors\n", - func, line, what, mv->msgcnt, msg_base, msg_base + exp_cnt, - exp_cnt); - else - TEST_SAY( - "%s:%d: %s: Verification of %d received messages " - "succeeded: " - "expected msgids %d..%d (%d)\n", - func, line, what, mv->msgcnt, msg_base, msg_base + exp_cnt, - exp_cnt); - - return fails; -} - -/** - * Verify that \p exp_cnt messages were received starting at - * msgid base \p msg_base. - */ -int test_msgver_verify0(const char *func, - int line, - const char *what, - test_msgver_t *mv, - int flags, - struct test_mv_vs vs) { - int fails = 0; - - TEST_SAY( - "%s:%d: %s: Verifying %d received messages (flags 0x%x): " - "expecting msgids %d..%d (%d)\n", - func, line, what, mv->msgcnt, flags, vs.msg_base, - vs.msg_base + vs.exp_cnt, vs.exp_cnt); - if (flags & TEST_MSGVER_BY_TIMESTAMP) { - assert((flags & TEST_MSGVER_BY_MSGID)); /* Required */ - TEST_SAY( - "%s:%d: %s: " - " and expecting timestamps %" PRId64 "..%" PRId64 "\n", - func, line, what, vs.timestamp_min, vs.timestamp_max); - } - - /* Per-partition checks */ - if (flags & TEST_MSGVER_ORDER) - fails += test_mv_p_verify_f(mv, flags, - test_mv_mvec_verify_order, &vs); - if (flags & TEST_MSGVER_DUP) - fails += - test_mv_p_verify_f(mv, flags, test_mv_mvec_verify_dup, &vs); - - if (flags & TEST_MSGVER_BY_BROKER_ID) - fails += test_mv_p_verify_f(mv, flags, - test_mv_mvec_verify_broker, &vs); - - /* Checks across all partitions */ - if ((flags & TEST_MSGVER_RANGE) && vs.exp_cnt > 0) { - vs.msgid_min = vs.msg_base; - vs.msgid_max = vs.msgid_min + vs.exp_cnt - 1; - fails += test_msgver_verify_range(mv, flags, &vs); - } - - if (mv->log_suppr_cnt > 0) - TEST_WARN("%s:%d: %s: %d message warning logs suppressed\n", - func, line, what, mv->log_suppr_cnt); - - if (vs.exp_cnt != mv->msgcnt) { - if (!(flags & TEST_MSGVER_SUBSET)) { - TEST_WARN("%s:%d: %s: expected %d messages, got %d\n", - func, line, what, vs.exp_cnt, mv->msgcnt); - fails++; - } - } - - if (fails) - TEST_FAIL( - "%s:%d: %s: Verification of %d received messages " - "failed: " - "expected msgids %d..%d (%d): see previous errors\n", - func, line, what, mv->msgcnt, vs.msg_base, - vs.msg_base + vs.exp_cnt, vs.exp_cnt); - else - TEST_SAY( - "%s:%d: %s: Verification of %d received messages " - "succeeded: " - "expected msgids %d..%d (%d)\n", - func, line, what, mv->msgcnt, vs.msg_base, - vs.msg_base + vs.exp_cnt, vs.exp_cnt); - - return fails; -} - - - -void test_verify_rkmessage0(const char *func, - int line, - rd_kafka_message_t *rkmessage, - uint64_t testid, - int32_t partition, - int msgnum) { - uint64_t in_testid; - int in_part; - int in_msgnum; - char buf[128]; - - rd_snprintf(buf, sizeof(buf), "%.*s", (int)rkmessage->len, - (char *)rkmessage->payload); - - if (sscanf(buf, "testid=%" SCNu64 ", partition=%i, msg=%i\n", - &in_testid, &in_part, &in_msgnum) != 3) - TEST_FAIL("Incorrect format: %s", buf); - - if (testid != in_testid || (partition != -1 && partition != in_part) || - (msgnum != -1 && msgnum != in_msgnum) || in_msgnum < 0) - goto fail_match; - - if (test_level > 2) { - TEST_SAY("%s:%i: Our testid %" PRIu64 - ", part %i (%i), msg %i\n", - func, line, testid, (int)partition, - (int)rkmessage->partition, msgnum); - } - - - return; - -fail_match: - TEST_FAIL("%s:%i: Our testid %" PRIu64 - ", part %i, msg %i did " - "not match message: \"%s\"\n", - func, line, testid, (int)partition, msgnum, buf); -} - - -/** - * @brief Verify that \p mv is identical to \p corr according to flags. - */ -void test_msgver_verify_compare0(const char *func, - int line, - const char *what, - test_msgver_t *mv, - test_msgver_t *corr, - int flags) { - struct test_mv_vs vs; - int fails = 0; - - memset(&vs, 0, sizeof(vs)); - - TEST_SAY( - "%s:%d: %s: Verifying %d received messages (flags 0x%x) by " - "comparison to correct msgver (%d messages)\n", - func, line, what, mv->msgcnt, flags, corr->msgcnt); - - vs.corr = corr; - - /* Per-partition checks */ - fails += test_mv_p_verify_f(mv, flags, test_mv_mvec_verify_corr, &vs); - - if (mv->log_suppr_cnt > 0) - TEST_WARN("%s:%d: %s: %d message warning logs suppressed\n", - func, line, what, mv->log_suppr_cnt); - - if (corr->msgcnt != mv->msgcnt) { - if (!(flags & TEST_MSGVER_SUBSET)) { - TEST_WARN("%s:%d: %s: expected %d messages, got %d\n", - func, line, what, corr->msgcnt, mv->msgcnt); - fails++; - } - } - - if (fails) - TEST_FAIL( - "%s:%d: %s: Verification of %d received messages " - "failed: expected %d messages: see previous errors\n", - func, line, what, mv->msgcnt, corr->msgcnt); - else - TEST_SAY( - "%s:%d: %s: Verification of %d received messages " - "succeeded: matching %d messages from correct msgver\n", - func, line, what, mv->msgcnt, corr->msgcnt); -} - - -/** - * Consumer poll but dont expect any proper messages for \p timeout_ms. - */ -void test_consumer_poll_no_msgs(const char *what, - rd_kafka_t *rk, - uint64_t testid, - int timeout_ms) { - int64_t tmout = test_clock() + ((int64_t)timeout_ms * 1000); - int cnt = 0; - test_timing_t t_cons; - test_msgver_t mv; - - test_msgver_init(&mv, testid); - - if (what) - TEST_SAY("%s: not expecting any messages for %dms\n", what, - timeout_ms); - - TIMING_START(&t_cons, "CONSUME"); - - do { - rd_kafka_message_t *rkmessage; - - rkmessage = rd_kafka_consumer_poll(rk, timeout_ms); - if (!rkmessage) - continue; - - if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { - TEST_SAY("%s [%" PRId32 - "] reached EOF at " - "offset %" PRId64 "\n", - rd_kafka_topic_name(rkmessage->rkt), - rkmessage->partition, rkmessage->offset); - test_msgver_add_msg(rk, &mv, rkmessage); - - } else if (rkmessage->err) { - TEST_FAIL( - "%s [%" PRId32 "] error (offset %" PRId64 "): %s", - rkmessage->rkt ? rd_kafka_topic_name(rkmessage->rkt) - : "(no-topic)", - rkmessage->partition, rkmessage->offset, - rd_kafka_message_errstr(rkmessage)); - - } else { - if (test_msgver_add_msg(rk, &mv, rkmessage)) { - TEST_MV_WARN( - &mv, - "Received unexpected message on " - "%s [%" PRId32 - "] at offset " - "%" PRId64 "\n", - rd_kafka_topic_name(rkmessage->rkt), - rkmessage->partition, rkmessage->offset); - cnt++; - } - } - - rd_kafka_message_destroy(rkmessage); - } while (test_clock() <= tmout); - - if (what) - TIMING_STOP(&t_cons); - - test_msgver_verify(what, &mv, TEST_MSGVER_ALL, 0, 0); - test_msgver_clear(&mv); - - TEST_ASSERT(cnt == 0, "Expected 0 messages, got %d", cnt); -} - -/** - * @brief Consumer poll with expectation that a \p err will be reached - * within \p timeout_ms. - */ -void test_consumer_poll_expect_err(rd_kafka_t *rk, - uint64_t testid, - int timeout_ms, - rd_kafka_resp_err_t err) { - int64_t tmout = test_clock() + ((int64_t)timeout_ms * 1000); - - TEST_SAY("%s: expecting error %s within %dms\n", rd_kafka_name(rk), - rd_kafka_err2name(err), timeout_ms); - - do { - rd_kafka_message_t *rkmessage; - rkmessage = rd_kafka_consumer_poll(rk, timeout_ms); - if (!rkmessage) - continue; - - if (rkmessage->err == err) { - TEST_SAY("Got expected error: %s: %s\n", - rd_kafka_err2name(rkmessage->err), - rd_kafka_message_errstr(rkmessage)); - rd_kafka_message_destroy(rkmessage); - - return; - } else if (rkmessage->err) { - TEST_FAIL("%s [%" PRId32 - "] unexpected error " - "(offset %" PRId64 "): %s", - rkmessage->rkt - ? rd_kafka_topic_name(rkmessage->rkt) - : "(no-topic)", - rkmessage->partition, rkmessage->offset, - rd_kafka_err2name(rkmessage->err)); - } - - rd_kafka_message_destroy(rkmessage); - } while (test_clock() <= tmout); - TEST_FAIL("Expected error %s not seen in %dms", rd_kafka_err2name(err), - timeout_ms); -} - -/** - * Call consumer poll once and then return. - * Messages are handled. - * - * \p mv is optional - * - * @returns 0 on timeout, 1 if a message was received or .._PARTITION_EOF - * if EOF was reached. - * TEST_FAIL()s on all errors. - */ -int test_consumer_poll_once(rd_kafka_t *rk, test_msgver_t *mv, int timeout_ms) { - rd_kafka_message_t *rkmessage; - - rkmessage = rd_kafka_consumer_poll(rk, timeout_ms); - if (!rkmessage) - return 0; - - if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { - TEST_SAY("%s [%" PRId32 - "] reached EOF at " - "offset %" PRId64 "\n", - rd_kafka_topic_name(rkmessage->rkt), - rkmessage->partition, rkmessage->offset); - if (mv) - test_msgver_add_msg(rk, mv, rkmessage); - rd_kafka_message_destroy(rkmessage); - return RD_KAFKA_RESP_ERR__PARTITION_EOF; - - } else if (rkmessage->err) { - TEST_FAIL("%s [%" PRId32 "] error (offset %" PRId64 "): %s", - rkmessage->rkt ? rd_kafka_topic_name(rkmessage->rkt) - : "(no-topic)", - rkmessage->partition, rkmessage->offset, - rd_kafka_message_errstr(rkmessage)); - - } else { - if (mv) - test_msgver_add_msg(rk, mv, rkmessage); - } - - rd_kafka_message_destroy(rkmessage); - return 1; -} - -/** - * @param exact Require exact exp_eof_cnt (unless -1) and exp_cnt (unless -1). - * If false: poll until either one is reached. - * @param timeout_ms Each call to poll has a timeout set by this argument. The - * test fails if any poll times out. - */ -int test_consumer_poll_exact_timeout(const char *what, - rd_kafka_t *rk, - uint64_t testid, - int exp_eof_cnt, - int exp_msg_base, - int exp_cnt, - rd_bool_t exact, - test_msgver_t *mv, - int timeout_ms) { - int eof_cnt = 0; - int cnt = 0; - test_timing_t t_cons; - - TEST_SAY("%s: consume %s%d messages\n", what, exact ? "exactly " : "", - exp_cnt); - - TIMING_START(&t_cons, "CONSUME"); - - while ((!exact && ((exp_eof_cnt <= 0 || eof_cnt < exp_eof_cnt) && - (exp_cnt <= 0 || cnt < exp_cnt))) || - (exact && (eof_cnt < exp_eof_cnt || cnt < exp_cnt))) { - rd_kafka_message_t *rkmessage; - - rkmessage = - rd_kafka_consumer_poll(rk, tmout_multip(timeout_ms)); - if (!rkmessage) /* Shouldn't take this long to get a msg */ - TEST_FAIL( - "%s: consumer_poll() timeout " - "(%d/%d eof, %d/%d msgs)\n", - what, eof_cnt, exp_eof_cnt, cnt, exp_cnt); - - - if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { - TEST_SAY("%s [%" PRId32 - "] reached EOF at " - "offset %" PRId64 "\n", - rd_kafka_topic_name(rkmessage->rkt), - rkmessage->partition, rkmessage->offset); - TEST_ASSERT(exp_eof_cnt != 0, "expected no EOFs"); - if (mv) - test_msgver_add_msg(rk, mv, rkmessage); - eof_cnt++; - - } else if (rkmessage->err) { - TEST_FAIL( - "%s [%" PRId32 "] error (offset %" PRId64 "): %s", - rkmessage->rkt ? rd_kafka_topic_name(rkmessage->rkt) - : "(no-topic)", - rkmessage->partition, rkmessage->offset, - rd_kafka_message_errstr(rkmessage)); - - } else { - TEST_SAYL(4, - "%s: consumed message on %s [%" PRId32 - "] " - "at offset %" PRId64 " (leader epoch %" PRId32 - ")\n", - what, rd_kafka_topic_name(rkmessage->rkt), - rkmessage->partition, rkmessage->offset, - rd_kafka_message_leader_epoch(rkmessage)); - - if (!mv || test_msgver_add_msg(rk, mv, rkmessage)) - cnt++; - } - - rd_kafka_message_destroy(rkmessage); - } - - TIMING_STOP(&t_cons); - - TEST_SAY("%s: consumed %d/%d messages (%d/%d EOFs)\n", what, cnt, - exp_cnt, eof_cnt, exp_eof_cnt); - - TEST_ASSERT(!exact || ((exp_cnt == -1 || exp_cnt == cnt) && - (exp_eof_cnt == -1 || exp_eof_cnt == eof_cnt)), - "%s: mismatch between exact expected counts and actual: " - "%d/%d EOFs, %d/%d msgs", - what, eof_cnt, exp_eof_cnt, cnt, exp_cnt); - - if (exp_cnt == 0) - TEST_ASSERT(cnt == 0 && eof_cnt == exp_eof_cnt, - "%s: expected no messages and %d EOFs: " - "got %d messages and %d EOFs", - what, exp_eof_cnt, cnt, eof_cnt); - return cnt; -} - - -/** - * @param exact Require exact exp_eof_cnt (unless -1) and exp_cnt (unless -1). - * If false: poll until either one is reached. - */ -int test_consumer_poll_exact(const char *what, - rd_kafka_t *rk, - uint64_t testid, - int exp_eof_cnt, - int exp_msg_base, - int exp_cnt, - rd_bool_t exact, - test_msgver_t *mv) { - return test_consumer_poll_exact_timeout(what, rk, testid, exp_eof_cnt, - exp_msg_base, exp_cnt, exact, - mv, 10 * 1000); -} - -int test_consumer_poll(const char *what, - rd_kafka_t *rk, - uint64_t testid, - int exp_eof_cnt, - int exp_msg_base, - int exp_cnt, - test_msgver_t *mv) { - return test_consumer_poll_exact(what, rk, testid, exp_eof_cnt, - exp_msg_base, exp_cnt, - rd_false /*not exact */, mv); -} - -int test_consumer_poll_timeout(const char *what, - rd_kafka_t *rk, - uint64_t testid, - int exp_eof_cnt, - int exp_msg_base, - int exp_cnt, - test_msgver_t *mv, - int timeout_ms) { - return test_consumer_poll_exact_timeout( - what, rk, testid, exp_eof_cnt, exp_msg_base, exp_cnt, - rd_false /*not exact */, mv, timeout_ms); -} - -void test_consumer_close(rd_kafka_t *rk) { - rd_kafka_resp_err_t err; - test_timing_t timing; - - TEST_SAY("Closing consumer %s\n", rd_kafka_name(rk)); - - TIMING_START(&timing, "CONSUMER.CLOSE"); - err = rd_kafka_consumer_close(rk); - TIMING_STOP(&timing); - if (err) - TEST_FAIL("Failed to close consumer: %s\n", - rd_kafka_err2str(err)); -} - - -void test_flush(rd_kafka_t *rk, int timeout_ms) { - test_timing_t timing; - rd_kafka_resp_err_t err; - - TEST_SAY("%s: Flushing %d messages\n", rd_kafka_name(rk), - rd_kafka_outq_len(rk)); - TIMING_START(&timing, "FLUSH"); - err = rd_kafka_flush(rk, timeout_ms); - TIMING_STOP(&timing); - if (err) - TEST_FAIL("Failed to flush(%s, %d): %s: len() = %d\n", - rd_kafka_name(rk), timeout_ms, rd_kafka_err2str(err), - rd_kafka_outq_len(rk)); -} - - -void test_conf_set(rd_kafka_conf_t *conf, const char *name, const char *val) { - char errstr[512]; - if (rd_kafka_conf_set(conf, name, val, errstr, sizeof(errstr)) != - RD_KAFKA_CONF_OK) - TEST_FAIL("Failed to set config \"%s\"=\"%s\": %s\n", name, val, - errstr); -} - -/** - * @brief Get configuration value for property \p name. - * - * @param conf Configuration to get value from. If NULL the test.conf (if any) - * configuration will be used. - */ -char *test_conf_get(const rd_kafka_conf_t *conf, const char *name) { - static RD_TLS char ret[256]; - size_t ret_sz = sizeof(ret); - rd_kafka_conf_t *def_conf = NULL; - - if (!conf) /* Use the current test.conf */ - test_conf_init(&def_conf, NULL, 0); - - if (rd_kafka_conf_get(conf ? conf : def_conf, name, ret, &ret_sz) != - RD_KAFKA_CONF_OK) - TEST_FAIL("Failed to get config \"%s\": %s\n", name, - "unknown property"); - - if (def_conf) - rd_kafka_conf_destroy(def_conf); - - return ret; -} - - -char *test_topic_conf_get(const rd_kafka_topic_conf_t *tconf, - const char *name) { - static RD_TLS char ret[256]; - size_t ret_sz = sizeof(ret); - if (rd_kafka_topic_conf_get(tconf, name, ret, &ret_sz) != - RD_KAFKA_CONF_OK) - TEST_FAIL("Failed to get topic config \"%s\": %s\n", name, - "unknown property"); - return ret; -} - - -/** - * @brief Check if property \name matches \p val in \p conf. - * If \p conf is NULL the test config will be used. */ -int test_conf_match(rd_kafka_conf_t *conf, const char *name, const char *val) { - char *real; - int free_conf = 0; - - if (!conf) { - test_conf_init(&conf, NULL, 0); - free_conf = 1; - } - - real = test_conf_get(conf, name); - - if (free_conf) - rd_kafka_conf_destroy(conf); - - return !strcmp(real, val); -} - - -void test_topic_conf_set(rd_kafka_topic_conf_t *tconf, - const char *name, - const char *val) { - char errstr[512]; - if (rd_kafka_topic_conf_set(tconf, name, val, errstr, sizeof(errstr)) != - RD_KAFKA_CONF_OK) - TEST_FAIL("Failed to set topic config \"%s\"=\"%s\": %s\n", - name, val, errstr); -} - -/** - * @brief First attempt to set topic level property, then global. - */ -void test_any_conf_set(rd_kafka_conf_t *conf, - rd_kafka_topic_conf_t *tconf, - const char *name, - const char *val) { - rd_kafka_conf_res_t res = RD_KAFKA_CONF_UNKNOWN; - char errstr[512] = {"Missing conf_t"}; - - if (tconf) - res = rd_kafka_topic_conf_set(tconf, name, val, errstr, - sizeof(errstr)); - if (res == RD_KAFKA_CONF_UNKNOWN && conf) - res = - rd_kafka_conf_set(conf, name, val, errstr, sizeof(errstr)); - - if (res != RD_KAFKA_CONF_OK) - TEST_FAIL("Failed to set any config \"%s\"=\"%s\": %s\n", name, - val, errstr); -} - - -/** - * @returns true if test clients need to be configured for authentication - * or other security measures (SSL), else false for unauthed plaintext. - */ -int test_needs_auth(void) { - rd_kafka_conf_t *conf; - const char *sec; - - test_conf_init(&conf, NULL, 0); - - sec = test_conf_get(conf, "security.protocol"); - - rd_kafka_conf_destroy(conf); - - return strcmp(sec, "plaintext"); -} - - -void test_print_partition_list( - const rd_kafka_topic_partition_list_t *partitions) { - int i; - for (i = 0; i < partitions->cnt; i++) { - TEST_SAY(" %s [%" PRId32 "] offset %" PRId64 " (epoch %" PRId32 - ") %s%s\n", - partitions->elems[i].topic, - partitions->elems[i].partition, - partitions->elems[i].offset, - rd_kafka_topic_partition_get_leader_epoch( - &partitions->elems[i]), - partitions->elems[i].err ? ": " : "", - partitions->elems[i].err - ? rd_kafka_err2str(partitions->elems[i].err) - : ""); - } -} - -/** - * @brief Compare two lists, returning 0 if equal. - * - * @remark The lists may be sorted by this function. - */ -int test_partition_list_cmp(rd_kafka_topic_partition_list_t *al, - rd_kafka_topic_partition_list_t *bl) { - int i; - - if (al->cnt < bl->cnt) - return -1; - else if (al->cnt > bl->cnt) - return 1; - else if (al->cnt == 0) - return 0; - - rd_kafka_topic_partition_list_sort(al, NULL, NULL); - rd_kafka_topic_partition_list_sort(bl, NULL, NULL); - - for (i = 0; i < al->cnt; i++) { - const rd_kafka_topic_partition_t *a = &al->elems[i]; - const rd_kafka_topic_partition_t *b = &bl->elems[i]; - if (a->partition != b->partition || strcmp(a->topic, b->topic)) - return -1; - } - - return 0; -} - -/** - * @brief Compare two lists and their offsets, returning 0 if equal. - * - * @remark The lists may be sorted by this function. - */ -int test_partition_list_and_offsets_cmp(rd_kafka_topic_partition_list_t *al, - rd_kafka_topic_partition_list_t *bl) { - int i; - - if (al->cnt < bl->cnt) - return -1; - else if (al->cnt > bl->cnt) - return 1; - else if (al->cnt == 0) - return 0; - - rd_kafka_topic_partition_list_sort(al, NULL, NULL); - rd_kafka_topic_partition_list_sort(bl, NULL, NULL); - - for (i = 0; i < al->cnt; i++) { - const rd_kafka_topic_partition_t *a = &al->elems[i]; - const rd_kafka_topic_partition_t *b = &bl->elems[i]; - if (a->partition != b->partition || - strcmp(a->topic, b->topic) || a->offset != b->offset || - rd_kafka_topic_partition_get_leader_epoch(a) != - rd_kafka_topic_partition_get_leader_epoch(b)) - return -1; - } - - return 0; -} - -/** - * @brief Execute script from the Kafka distribution bin/ path. - */ -void test_kafka_cmd(const char *fmt, ...) { -#ifdef _WIN32 - TEST_FAIL("%s not supported on Windows, yet", __FUNCTION__); -#else - char cmd[1024]; - int r; - va_list ap; - test_timing_t t_cmd; - const char *kpath; - - kpath = test_getenv("KAFKA_PATH", NULL); - - if (!kpath) - TEST_FAIL("%s: KAFKA_PATH must be set", __FUNCTION__); - - r = rd_snprintf(cmd, sizeof(cmd), "%s/bin/", kpath); - TEST_ASSERT(r < (int)sizeof(cmd)); - - va_start(ap, fmt); - rd_vsnprintf(cmd + r, sizeof(cmd) - r, fmt, ap); - va_end(ap); - - TEST_SAY("Executing: %s\n", cmd); - TIMING_START(&t_cmd, "exec"); - r = system(cmd); - TIMING_STOP(&t_cmd); - - if (r == -1) - TEST_FAIL("system(\"%s\") failed: %s", cmd, strerror(errno)); - else if (WIFSIGNALED(r)) - TEST_FAIL("system(\"%s\") terminated by signal %d\n", cmd, - WTERMSIG(r)); - else if (WEXITSTATUS(r)) - TEST_FAIL("system(\"%s\") failed with exit status %d\n", cmd, - WEXITSTATUS(r)); -#endif -} - -/** - * @brief Execute kafka-topics.sh from the Kafka distribution. - */ -void test_kafka_topics(const char *fmt, ...) { -#ifdef _WIN32 - TEST_FAIL("%s not supported on Windows, yet", __FUNCTION__); -#else - char cmd[1024]; - int r, bytes_left; - va_list ap; - test_timing_t t_cmd; - const char *kpath, *bootstrap_env, *flag, *bootstrap_srvs; - - if (test_broker_version >= TEST_BRKVER(3, 0, 0, 0)) { - bootstrap_env = "BROKERS"; - flag = "--bootstrap-server"; - } else { - bootstrap_env = "ZK_ADDRESS"; - flag = "--zookeeper"; - } - - kpath = test_getenv("KAFKA_PATH", NULL); - bootstrap_srvs = test_getenv(bootstrap_env, NULL); - - if (!kpath || !bootstrap_srvs) - TEST_FAIL("%s: KAFKA_PATH and %s must be set", __FUNCTION__, - bootstrap_env); - - r = rd_snprintf(cmd, sizeof(cmd), "%s/bin/kafka-topics.sh %s %s ", - kpath, flag, bootstrap_srvs); - TEST_ASSERT(r > 0 && r < (int)sizeof(cmd)); - - bytes_left = sizeof(cmd) - r; - - va_start(ap, fmt); - r = rd_vsnprintf(cmd + r, bytes_left, fmt, ap); - va_end(ap); - TEST_ASSERT(r > 0 && r < bytes_left); - - TEST_SAY("Executing: %s\n", cmd); - TIMING_START(&t_cmd, "exec"); - r = system(cmd); - TIMING_STOP(&t_cmd); - - if (r == -1) - TEST_FAIL("system(\"%s\") failed: %s", cmd, strerror(errno)); - else if (WIFSIGNALED(r)) - TEST_FAIL("system(\"%s\") terminated by signal %d\n", cmd, - WTERMSIG(r)); - else if (WEXITSTATUS(r)) - TEST_FAIL("system(\"%s\") failed with exit status %d\n", cmd, - WEXITSTATUS(r)); -#endif -} - - - -/** - * @brief Create topic using Topic Admin API - * - * @param configs is an optional key-value tuple array of - * topic configs (or NULL). - */ -void test_admin_create_topic(rd_kafka_t *use_rk, - const char *topicname, - int partition_cnt, - int replication_factor, - const char **configs) { - rd_kafka_t *rk; - rd_kafka_NewTopic_t *newt[1]; - const size_t newt_cnt = 1; - rd_kafka_AdminOptions_t *options; - rd_kafka_queue_t *rkqu; - rd_kafka_event_t *rkev; - const rd_kafka_CreateTopics_result_t *res; - const rd_kafka_topic_result_t **terr; - int timeout_ms = tmout_multip(10000); - size_t res_cnt; - rd_kafka_resp_err_t err; - char errstr[512]; - test_timing_t t_create; - - if (!(rk = use_rk)) - rk = test_create_producer(); - - rkqu = rd_kafka_queue_new(rk); - - newt[0] = - rd_kafka_NewTopic_new(topicname, partition_cnt, replication_factor, - errstr, sizeof(errstr)); - TEST_ASSERT(newt[0] != NULL, "%s", errstr); - - if (configs) { - int i; - - for (i = 0; configs[i] && configs[i + 1]; i += 2) - TEST_CALL_ERR__(rd_kafka_NewTopic_set_config( - newt[0], configs[i], configs[i + 1])); - } - - options = rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_CREATETOPICS); - err = rd_kafka_AdminOptions_set_operation_timeout( - options, timeout_ms, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "%s", errstr); - - TEST_SAY( - "Creating topic \"%s\" " - "(partitions=%d, replication_factor=%d, timeout=%d)\n", - topicname, partition_cnt, replication_factor, timeout_ms); - - TIMING_START(&t_create, "CreateTopics"); - rd_kafka_CreateTopics(rk, newt, newt_cnt, options, rkqu); - - /* Wait for result */ - rkev = rd_kafka_queue_poll(rkqu, timeout_ms + 2000); - TEST_ASSERT(rkev, "Timed out waiting for CreateTopics result"); - - TIMING_STOP(&t_create); - - TEST_ASSERT(!rd_kafka_event_error(rkev), "CreateTopics failed: %s", - rd_kafka_event_error_string(rkev)); - - res = rd_kafka_event_CreateTopics_result(rkev); - TEST_ASSERT(res, "Expected CreateTopics_result, not %s", - rd_kafka_event_name(rkev)); - - terr = rd_kafka_CreateTopics_result_topics(res, &res_cnt); - TEST_ASSERT(terr, "CreateTopics_result_topics returned NULL"); - TEST_ASSERT(res_cnt == newt_cnt, - "CreateTopics_result_topics returned %" PRIusz - " topics, " - "not the expected %" PRIusz, - res_cnt, newt_cnt); - - TEST_ASSERT(!rd_kafka_topic_result_error(terr[0]) || - rd_kafka_topic_result_error(terr[0]) == - RD_KAFKA_RESP_ERR_TOPIC_ALREADY_EXISTS, - "Topic %s result error: %s", - rd_kafka_topic_result_name(terr[0]), - rd_kafka_topic_result_error_string(terr[0])); - - rd_kafka_event_destroy(rkev); - - rd_kafka_queue_destroy(rkqu); - - rd_kafka_AdminOptions_destroy(options); - - rd_kafka_NewTopic_destroy(newt[0]); - - if (!use_rk) - rd_kafka_destroy(rk); -} - - - -/** - * @brief Create topic using kafka-topics.sh --create - */ -static void test_create_topic_sh(const char *topicname, - int partition_cnt, - int replication_factor) { - test_kafka_topics( - "--create --topic \"%s\" " - "--replication-factor %d --partitions %d", - topicname, replication_factor, partition_cnt); -} - - -/** - * @brief Create topic - */ -void test_create_topic(rd_kafka_t *use_rk, - const char *topicname, - int partition_cnt, - int replication_factor) { - if (test_broker_version < TEST_BRKVER(0, 10, 2, 0)) - test_create_topic_sh(topicname, partition_cnt, - replication_factor); - else - test_admin_create_topic(use_rk, topicname, partition_cnt, - replication_factor, NULL); -} - - -/** - * @brief Create topic using kafka-topics.sh --delete - */ -static void test_delete_topic_sh(const char *topicname) { - test_kafka_topics("--delete --topic \"%s\" ", topicname); -} - - -/** - * @brief Delete topic using Topic Admin API - */ -static void test_admin_delete_topic(rd_kafka_t *use_rk, const char *topicname) { - rd_kafka_t *rk; - rd_kafka_DeleteTopic_t *delt[1]; - const size_t delt_cnt = 1; - rd_kafka_AdminOptions_t *options; - rd_kafka_queue_t *rkqu; - rd_kafka_event_t *rkev; - const rd_kafka_DeleteTopics_result_t *res; - const rd_kafka_topic_result_t **terr; - int timeout_ms = tmout_multip(10000); - size_t res_cnt; - rd_kafka_resp_err_t err; - char errstr[512]; - test_timing_t t_create; - - if (!(rk = use_rk)) - rk = test_create_producer(); - - rkqu = rd_kafka_queue_new(rk); - - delt[0] = rd_kafka_DeleteTopic_new(topicname); - - options = rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_DELETETOPICS); - err = rd_kafka_AdminOptions_set_operation_timeout( - options, timeout_ms, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "%s", errstr); - - TEST_SAY( - "Deleting topic \"%s\" " - "(timeout=%d)\n", - topicname, timeout_ms); - - TIMING_START(&t_create, "DeleteTopics"); - rd_kafka_DeleteTopics(rk, delt, delt_cnt, options, rkqu); - - /* Wait for result */ - rkev = rd_kafka_queue_poll(rkqu, timeout_ms + 2000); - TEST_ASSERT(rkev, "Timed out waiting for DeleteTopics result"); - - TIMING_STOP(&t_create); - - res = rd_kafka_event_DeleteTopics_result(rkev); - TEST_ASSERT(res, "Expected DeleteTopics_result, not %s", - rd_kafka_event_name(rkev)); - - terr = rd_kafka_DeleteTopics_result_topics(res, &res_cnt); - TEST_ASSERT(terr, "DeleteTopics_result_topics returned NULL"); - TEST_ASSERT(res_cnt == delt_cnt, - "DeleteTopics_result_topics returned %" PRIusz - " topics, " - "not the expected %" PRIusz, - res_cnt, delt_cnt); - - TEST_ASSERT(!rd_kafka_topic_result_error(terr[0]), - "Topic %s result error: %s", - rd_kafka_topic_result_name(terr[0]), - rd_kafka_topic_result_error_string(terr[0])); - - rd_kafka_event_destroy(rkev); - - rd_kafka_queue_destroy(rkqu); - - rd_kafka_AdminOptions_destroy(options); - - rd_kafka_DeleteTopic_destroy(delt[0]); - - if (!use_rk) - rd_kafka_destroy(rk); -} - - -/** - * @brief Delete a topic - */ -void test_delete_topic(rd_kafka_t *use_rk, const char *topicname) { - if (test_broker_version < TEST_BRKVER(0, 10, 2, 0)) - test_delete_topic_sh(topicname); - else - test_admin_delete_topic(use_rk, topicname); -} - - -/** - * @brief Create additional partitions for a topic using Admin API - */ -static void test_admin_create_partitions(rd_kafka_t *use_rk, - const char *topicname, - int new_partition_cnt) { - rd_kafka_t *rk; - rd_kafka_NewPartitions_t *newp[1]; - const size_t newp_cnt = 1; - rd_kafka_AdminOptions_t *options; - rd_kafka_queue_t *rkqu; - rd_kafka_event_t *rkev; - const rd_kafka_CreatePartitions_result_t *res; - const rd_kafka_topic_result_t **terr; - int timeout_ms = tmout_multip(10000); - size_t res_cnt; - rd_kafka_resp_err_t err; - char errstr[512]; - test_timing_t t_create; - - if (!(rk = use_rk)) - rk = test_create_producer(); - - rkqu = rd_kafka_queue_new(rk); - - newp[0] = rd_kafka_NewPartitions_new(topicname, new_partition_cnt, - errstr, sizeof(errstr)); - TEST_ASSERT(newp[0] != NULL, "%s", errstr); - - options = - rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_CREATEPARTITIONS); - err = rd_kafka_AdminOptions_set_operation_timeout( - options, timeout_ms, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "%s", errstr); - - TEST_SAY("Creating %d (total) partitions for topic \"%s\"\n", - new_partition_cnt, topicname); - - TIMING_START(&t_create, "CreatePartitions"); - rd_kafka_CreatePartitions(rk, newp, newp_cnt, options, rkqu); - - /* Wait for result */ - rkev = rd_kafka_queue_poll(rkqu, timeout_ms + 2000); - TEST_ASSERT(rkev, "Timed out waiting for CreatePartitions result"); - - TIMING_STOP(&t_create); - - res = rd_kafka_event_CreatePartitions_result(rkev); - TEST_ASSERT(res, "Expected CreatePartitions_result, not %s", - rd_kafka_event_name(rkev)); - - terr = rd_kafka_CreatePartitions_result_topics(res, &res_cnt); - TEST_ASSERT(terr, "CreatePartitions_result_topics returned NULL"); - TEST_ASSERT(res_cnt == newp_cnt, - "CreatePartitions_result_topics returned %" PRIusz - " topics, not the expected %" PRIusz, - res_cnt, newp_cnt); - - TEST_ASSERT(!rd_kafka_topic_result_error(terr[0]), - "Topic %s result error: %s", - rd_kafka_topic_result_name(terr[0]), - rd_kafka_topic_result_error_string(terr[0])); - - rd_kafka_event_destroy(rkev); - - rd_kafka_queue_destroy(rkqu); - - rd_kafka_AdminOptions_destroy(options); - - rd_kafka_NewPartitions_destroy(newp[0]); - - if (!use_rk) - rd_kafka_destroy(rk); -} - - -/** - * @brief Create partitions for topic - */ -void test_create_partitions(rd_kafka_t *use_rk, - const char *topicname, - int new_partition_cnt) { - if (test_broker_version < TEST_BRKVER(0, 10, 2, 0)) - test_kafka_topics("--alter --topic %s --partitions %d", - topicname, new_partition_cnt); - else - test_admin_create_partitions(use_rk, topicname, - new_partition_cnt); -} - - -int test_get_partition_count(rd_kafka_t *rk, - const char *topicname, - int timeout_ms) { - rd_kafka_t *use_rk; - rd_kafka_resp_err_t err; - rd_kafka_topic_t *rkt; - int64_t abs_timeout = test_clock() + ((int64_t)timeout_ms * 1000); - int ret = -1; - - if (!rk) - use_rk = test_create_producer(); - else - use_rk = rk; - - rkt = rd_kafka_topic_new(use_rk, topicname, NULL); - - do { - const struct rd_kafka_metadata *metadata; - - err = rd_kafka_metadata(use_rk, 0, rkt, &metadata, - tmout_multip(15000)); - if (err) - TEST_WARN("metadata() for %s failed: %s\n", - rkt ? rd_kafka_topic_name(rkt) - : "(all-local)", - rd_kafka_err2str(err)); - else { - if (metadata->topic_cnt == 1) { - if (metadata->topics[0].err == 0 || - metadata->topics[0].partition_cnt > 0) { - int32_t cnt; - cnt = metadata->topics[0].partition_cnt; - rd_kafka_metadata_destroy(metadata); - ret = (int)cnt; - break; - } - TEST_SAY( - "metadata(%s) returned %s: retrying\n", - rd_kafka_topic_name(rkt), - rd_kafka_err2str(metadata->topics[0].err)); - } - rd_kafka_metadata_destroy(metadata); - rd_sleep(1); - } - } while (test_clock() < abs_timeout); - - rd_kafka_topic_destroy(rkt); - - if (!rk) - rd_kafka_destroy(use_rk); - - return ret; -} - -/** - * @brief Let the broker auto-create the topic for us. - */ -rd_kafka_resp_err_t test_auto_create_topic_rkt(rd_kafka_t *rk, - rd_kafka_topic_t *rkt, - int timeout_ms) { - const struct rd_kafka_metadata *metadata; - rd_kafka_resp_err_t err; - test_timing_t t; - int64_t abs_timeout = test_clock() + ((int64_t)timeout_ms * 1000); - - do { - TIMING_START(&t, "auto_create_topic"); - err = rd_kafka_metadata(rk, 0, rkt, &metadata, - tmout_multip(15000)); - TIMING_STOP(&t); - if (err) - TEST_WARN("metadata() for %s failed: %s\n", - rkt ? rd_kafka_topic_name(rkt) - : "(all-local)", - rd_kafka_err2str(err)); - else { - if (metadata->topic_cnt == 1) { - if (metadata->topics[0].err == 0 || - metadata->topics[0].partition_cnt > 0) { - rd_kafka_metadata_destroy(metadata); - return 0; - } - TEST_SAY( - "metadata(%s) returned %s: retrying\n", - rd_kafka_topic_name(rkt), - rd_kafka_err2str(metadata->topics[0].err)); - } - rd_kafka_metadata_destroy(metadata); - rd_sleep(1); - } - } while (test_clock() < abs_timeout); - - return err; -} - -rd_kafka_resp_err_t -test_auto_create_topic(rd_kafka_t *rk, const char *name, int timeout_ms) { - rd_kafka_topic_t *rkt = rd_kafka_topic_new(rk, name, NULL); - rd_kafka_resp_err_t err; - if (!rkt) - return rd_kafka_last_error(); - err = test_auto_create_topic_rkt(rk, rkt, timeout_ms); - rd_kafka_topic_destroy(rkt); - return err; -} - - -/** - * @brief Check if topic auto creation works. - * @returns 1 if it does, else 0. - */ -int test_check_auto_create_topic(void) { - rd_kafka_t *rk; - rd_kafka_conf_t *conf; - rd_kafka_resp_err_t err; - const char *topic = test_mk_topic_name("autocreatetest", 1); - - test_conf_init(&conf, NULL, 0); - rk = test_create_handle(RD_KAFKA_PRODUCER, conf); - err = test_auto_create_topic(rk, topic, tmout_multip(5000)); - if (err) - TEST_SAY("Auto topic creation of \"%s\" failed: %s\n", topic, - rd_kafka_err2str(err)); - rd_kafka_destroy(rk); - - return err ? 0 : 1; -} - - -/** - * @brief Builds and runs a Java application from the java/ directory. - * - * The application is started in the background, use - * test_waitpid() to await its demise. - * - * @param cls The app class to run using java/run-class.sh - * - * @returns -1 if the application could not be started, else the pid. - */ -int test_run_java(const char *cls, const char **argv) { -#ifdef _WIN32 - TEST_WARN("%s(%s) not supported Windows, yet", __FUNCTION__, cls); - return -1; -#else - int r; - const char *kpath; - pid_t pid; - const char **full_argv, **p; - int cnt; - extern char **environ; - - kpath = test_getenv("KAFKA_PATH", NULL); - - if (!kpath) { - TEST_WARN("%s(%s): KAFKA_PATH must be set\n", __FUNCTION__, - cls); - return -1; - } - - /* Build */ - r = system("make -s java"); - - if (r == -1 || WIFSIGNALED(r) || WEXITSTATUS(r)) { - TEST_WARN("%s(%s): failed to build java class (code %d)\n", - __FUNCTION__, cls, r); - return -1; - } - - /* For child process and run cls */ - pid = fork(); - if (pid == -1) { - TEST_WARN("%s(%s): failed to fork: %s\n", __FUNCTION__, cls, - strerror(errno)); - return -1; - } - - if (pid > 0) - return (int)pid; /* In parent process */ - - /* In child process */ - - /* Reconstruct argv to contain run-class.sh and the cls */ - for (cnt = 0; argv[cnt]; cnt++) - ; - - cnt += 3; /* run-class.sh, cls, .., NULL */ - full_argv = malloc(sizeof(*full_argv) * cnt); - full_argv[0] = "java/run-class.sh"; - full_argv[1] = (const char *)cls; - - /* Copy arguments */ - for (p = &full_argv[2]; *argv; p++, argv++) - *p = *argv; - *p = NULL; - - /* Run */ - r = execve(full_argv[0], (char *const *)full_argv, environ); - - TEST_WARN("%s(%s): failed to execute run-class.sh: %s\n", __FUNCTION__, - cls, strerror(errno)); - exit(2); - - return -1; /* NOTREACHED */ -#endif -} - - -/** - * @brief Wait for child-process \p pid to exit. - * - * @returns -1 if the child process exited successfully, else -1. - */ -int test_waitpid(int pid) { -#ifdef _WIN32 - TEST_WARN("%s() not supported Windows, yet", __FUNCTION__); - return -1; -#else - pid_t r; - int status = 0; - - r = waitpid((pid_t)pid, &status, 0); - - if (r == -1) { - TEST_WARN("waitpid(%d) failed: %s\n", pid, strerror(errno)); - return -1; - } - - if (WIFSIGNALED(status)) { - TEST_WARN("Process %d terminated by signal %d\n", pid, - WTERMSIG(status)); - return -1; - } else if (WEXITSTATUS(status)) { - TEST_WARN("Process %d exited with status %d\n", pid, - WEXITSTATUS(status)); - return -1; - } - - return 0; -#endif -} - - -/** - * @brief Check if \p feature is builtin to librdkafka. - * @returns returns 1 if feature is built in, else 0. - */ -int test_check_builtin(const char *feature) { - rd_kafka_conf_t *conf; - char errstr[128]; - int r; - - conf = rd_kafka_conf_new(); - if (rd_kafka_conf_set(conf, "builtin.features", feature, errstr, - sizeof(errstr)) != RD_KAFKA_CONF_OK) { - TEST_SAY("Feature \"%s\" not built-in: %s\n", feature, errstr); - r = 0; - } else { - TEST_SAY("Feature \"%s\" is built-in\n", feature); - r = 1; - } - - rd_kafka_conf_destroy(conf); - return r; -} - - -char *tsprintf(const char *fmt, ...) { - static RD_TLS char ret[8][512]; - static RD_TLS int i; - va_list ap; - - - i = (i + 1) % 8; - - va_start(ap, fmt); - rd_vsnprintf(ret[i], sizeof(ret[i]), fmt, ap); - va_end(ap); - - return ret[i]; -} - - -/** - * @brief Add a test report JSON object. - * These will be written as a JSON array to the test report file. - */ -void test_report_add(struct test *test, const char *fmt, ...) { - va_list ap; - char buf[512]; - - va_start(ap, fmt); - vsnprintf(buf, sizeof(buf), fmt, ap); - va_end(ap); - - if (test->report_cnt == test->report_size) { - if (test->report_size == 0) - test->report_size = 8; - else - test->report_size *= 2; - - test->report_arr = - realloc(test->report_arr, - sizeof(*test->report_arr) * test->report_size); - } - - test->report_arr[test->report_cnt++] = rd_strdup(buf); - - TEST_SAYL(1, "Report #%d: %s\n", test->report_cnt - 1, buf); -} - -/** - * Returns 1 if KAFKA_PATH and BROKERS (or ZK_ADDRESS) is set to se we can use - * the kafka-topics.sh script to manually create topics. - * - * If \p skip is set TEST_SKIP() will be called with a helpful message. - */ -int test_can_create_topics(int skip) { -#ifndef _WIN32 - const char *bootstrap; -#endif - - /* Has AdminAPI */ - if (test_broker_version >= TEST_BRKVER(0, 10, 2, 0)) - return 1; - -#ifdef _WIN32 - if (skip) - TEST_SKIP("Cannot create topics on Win32\n"); - return 0; -#else - - bootstrap = test_broker_version >= TEST_BRKVER(3, 0, 0, 0) - ? "BROKERS" - : "ZK_ADDRESS"; - - if (!test_getenv("KAFKA_PATH", NULL) || !test_getenv(bootstrap, NULL)) { - if (skip) - TEST_SKIP( - "Cannot create topics " - "(set KAFKA_PATH and %s)\n", - bootstrap); - return 0; - } - - - return 1; -#endif -} - - -/** - * Wait for \p event_type, discarding all other events prior to it. - */ -rd_kafka_event_t *test_wait_event(rd_kafka_queue_t *eventq, - rd_kafka_event_type_t event_type, - int timeout_ms) { - test_timing_t t_w; - int64_t abs_timeout = test_clock() + ((int64_t)timeout_ms * 1000); - - TIMING_START(&t_w, "wait_event"); - while (test_clock() < abs_timeout) { - rd_kafka_event_t *rkev; - - rkev = rd_kafka_queue_poll( - eventq, (int)(abs_timeout - test_clock()) / 1000); - - if (rd_kafka_event_type(rkev) == event_type) { - TIMING_STOP(&t_w); - return rkev; - } - - if (!rkev) - continue; - - if (rd_kafka_event_error(rkev)) - TEST_SAY("discarding ignored event %s: %s\n", - rd_kafka_event_name(rkev), - rd_kafka_event_error_string(rkev)); - else - TEST_SAY("discarding ignored event %s\n", - rd_kafka_event_name(rkev)); - rd_kafka_event_destroy(rkev); - } - TIMING_STOP(&t_w); - - return NULL; -} - - -void test_SAY(const char *file, int line, int level, const char *str) { - TEST_SAYL(level, "%s", str); -} - -void test_SKIP(const char *file, int line, const char *str) { - TEST_WARN("SKIPPING TEST: %s", str); - TEST_LOCK(); - test_curr->state = TEST_SKIPPED; - if (!*test_curr->failstr) { - rd_snprintf(test_curr->failstr, sizeof(test_curr->failstr), - "%s", str); - rtrim(test_curr->failstr); - } - TEST_UNLOCK(); -} - -const char *test_curr_name(void) { - return test_curr->name; -} - - -/** - * @brief Dump/print message haders - */ -void test_headers_dump(const char *what, - int lvl, - const rd_kafka_headers_t *hdrs) { - size_t idx = 0; - const char *name, *value; - size_t size; - - while (!rd_kafka_header_get_all(hdrs, idx++, &name, - (const void **)&value, &size)) - TEST_SAYL(lvl, "%s: Header #%" PRIusz ": %s='%s'\n", what, - idx - 1, name, value ? value : "(NULL)"); -} - - -/** - * @brief Retrieve and return the list of broker ids in the cluster. - * - * @param rk Optional instance to use. - * @param cntp Will be updated to the number of brokers returned. - * - * @returns a malloc:ed list of int32_t broker ids. - */ -int32_t *test_get_broker_ids(rd_kafka_t *use_rk, size_t *cntp) { - int32_t *ids; - rd_kafka_t *rk; - const rd_kafka_metadata_t *md; - rd_kafka_resp_err_t err; - size_t i; - - if (!(rk = use_rk)) - rk = test_create_producer(); - - err = rd_kafka_metadata(rk, 0, NULL, &md, tmout_multip(5000)); - TEST_ASSERT(!err, "%s", rd_kafka_err2str(err)); - TEST_ASSERT(md->broker_cnt > 0, "%d brokers, expected > 0", - md->broker_cnt); - - ids = malloc(sizeof(*ids) * md->broker_cnt); - - for (i = 0; i < (size_t)md->broker_cnt; i++) - ids[i] = md->brokers[i].id; - - *cntp = md->broker_cnt; - - rd_kafka_metadata_destroy(md); - - if (!use_rk) - rd_kafka_destroy(rk); - - return ids; -} - - - -/** - * @brief Verify that all topics in \p topics are reported in metadata, - * and that none of the topics in \p not_topics are reported. - * - * @returns the number of failures (but does not FAIL). - */ -static int verify_topics_in_metadata(rd_kafka_t *rk, - rd_kafka_metadata_topic_t *topics, - size_t topic_cnt, - rd_kafka_metadata_topic_t *not_topics, - size_t not_topic_cnt) { - const rd_kafka_metadata_t *md; - rd_kafka_resp_err_t err; - int ti; - size_t i; - int fails = 0; - - /* Mark topics with dummy error which is overwritten - * when topic is found in metadata, allowing us to check - * for missed topics. */ - for (i = 0; i < topic_cnt; i++) - topics[i].err = 12345; - - err = rd_kafka_metadata(rk, 1 /*all_topics*/, NULL, &md, - tmout_multip(5000)); - TEST_ASSERT(!err, "metadata failed: %s", rd_kafka_err2str(err)); - - for (ti = 0; ti < md->topic_cnt; ti++) { - const rd_kafka_metadata_topic_t *mdt = &md->topics[ti]; - - for (i = 0; i < topic_cnt; i++) { - int pi; - rd_kafka_metadata_topic_t *exp_mdt; - - if (strcmp(topics[i].topic, mdt->topic)) - continue; - - exp_mdt = &topics[i]; - - exp_mdt->err = mdt->err; /* indicate found */ - if (mdt->err) { - TEST_SAY( - "metadata: " - "Topic %s has error %s\n", - mdt->topic, rd_kafka_err2str(mdt->err)); - fails++; - } - - if (exp_mdt->partition_cnt > 0 && - mdt->partition_cnt != exp_mdt->partition_cnt) { - TEST_SAY( - "metadata: " - "Topic %s, expected %d partitions" - ", not %d\n", - mdt->topic, exp_mdt->partition_cnt, - mdt->partition_cnt); - fails++; - continue; - } - - /* Verify per-partition values */ - for (pi = 0; - exp_mdt->partitions && pi < exp_mdt->partition_cnt; - pi++) { - const rd_kafka_metadata_partition_t *mdp = - &mdt->partitions[pi]; - const rd_kafka_metadata_partition_t *exp_mdp = - &exp_mdt->partitions[pi]; - - if (mdp->id != exp_mdp->id) { - TEST_SAY( - "metadata: " - "Topic %s, " - "partition %d, " - "partition list out of order," - " expected %d, not %d\n", - mdt->topic, pi, exp_mdp->id, - mdp->id); - fails++; - continue; - } - - if (exp_mdp->replicas) { - if (mdp->replica_cnt != - exp_mdp->replica_cnt) { - TEST_SAY( - "metadata: " - "Topic %s, " - "partition %d, " - "expected %d replicas," - " not %d\n", - mdt->topic, pi, - exp_mdp->replica_cnt, - mdp->replica_cnt); - fails++; - } else if ( - memcmp( - mdp->replicas, - exp_mdp->replicas, - mdp->replica_cnt * - sizeof(*mdp->replicas))) { - int ri; - - TEST_SAY( - "metadata: " - "Topic %s, " - "partition %d, " - "replica mismatch:\n", - mdt->topic, pi); - - for (ri = 0; - ri < mdp->replica_cnt; - ri++) { - TEST_SAY( - " #%d: " - "expected " - "replica %d, " - "not %d\n", - ri, - exp_mdp - ->replicas[ri], - mdp->replicas[ri]); - } - - fails++; - } - } - } - } - - for (i = 0; i < not_topic_cnt; i++) { - if (strcmp(not_topics[i].topic, mdt->topic)) - continue; - - TEST_SAY( - "metadata: " - "Topic %s found in metadata, unexpected\n", - mdt->topic); - fails++; - } - } - - for (i = 0; i < topic_cnt; i++) { - if ((int)topics[i].err == 12345) { - TEST_SAY( - "metadata: " - "Topic %s not seen in metadata\n", - topics[i].topic); - fails++; - } - } - - if (fails > 0) - TEST_SAY("Metadata verification for %" PRIusz - " topics failed " - "with %d errors (see above)\n", - topic_cnt, fails); - else - TEST_SAY( - "Metadata verification succeeded: " - "%" PRIusz - " desired topics seen, " - "%" PRIusz " undesired topics not seen\n", - topic_cnt, not_topic_cnt); - - rd_kafka_metadata_destroy(md); - - return fails; -} - - - -/** - * @brief Wait for metadata to reflect expected and not expected topics - */ -void test_wait_metadata_update(rd_kafka_t *rk, - rd_kafka_metadata_topic_t *topics, - size_t topic_cnt, - rd_kafka_metadata_topic_t *not_topics, - size_t not_topic_cnt, - int tmout) { - int64_t abs_timeout; - test_timing_t t_md; - rd_kafka_t *our_rk = NULL; - - if (!rk) - rk = our_rk = test_create_handle(RD_KAFKA_PRODUCER, NULL); - - abs_timeout = test_clock() + ((int64_t)tmout * 1000); - - TEST_SAY("Waiting for up to %dms for metadata update\n", tmout); - - TIMING_START(&t_md, "METADATA.WAIT"); - do { - int md_fails; - - md_fails = verify_topics_in_metadata(rk, topics, topic_cnt, - not_topics, not_topic_cnt); - - if (!md_fails) { - TEST_SAY( - "All expected topics (not?) " - "seen in metadata\n"); - abs_timeout = 0; - break; - } - - rd_sleep(1); - } while (test_clock() < abs_timeout); - TIMING_STOP(&t_md); - - if (our_rk) - rd_kafka_destroy(our_rk); - - if (abs_timeout) - TEST_FAIL("Expected topics not seen in given time."); -} - -/** - * @brief Wait for topic to be available in metadata - */ -void test_wait_topic_exists(rd_kafka_t *rk, const char *topic, int tmout) { - rd_kafka_metadata_topic_t topics = {.topic = (char *)topic}; - - test_wait_metadata_update(rk, &topics, 1, NULL, 0, tmout); - - /* Wait an additional second for the topic to propagate in - * the cluster. This is not perfect but a cheap workaround for - * the asynchronous nature of topic creations in Kafka. */ - rd_sleep(1); -} - - - -/** - * @brief Wait for up to \p tmout for any type of admin result. - * @returns the event - */ -rd_kafka_event_t *test_wait_admin_result(rd_kafka_queue_t *q, - rd_kafka_event_type_t evtype, - int tmout) { - rd_kafka_event_t *rkev; - - while (1) { - rkev = rd_kafka_queue_poll(q, tmout); - if (!rkev) - TEST_FAIL("Timed out waiting for admin result (%d)\n", - evtype); - - if (rd_kafka_event_type(rkev) == evtype) - return rkev; - - - if (rd_kafka_event_type(rkev) == RD_KAFKA_EVENT_ERROR) { - TEST_WARN( - "Received error event while waiting for %d: " - "%s: ignoring", - evtype, rd_kafka_event_error_string(rkev)); - continue; - } - - - TEST_ASSERT(rd_kafka_event_type(rkev) == evtype, - "Expected event type %d, got %d (%s)", evtype, - rd_kafka_event_type(rkev), - rd_kafka_event_name(rkev)); - } - - return NULL; -} - -/** - * @brief Wait for up to \p tmout for an admin API result and return the - * distilled error code. - * - * Supported APIs: - * - AlterConfigs - * - CreatePartitions - * - CreateTopics - * - DeleteGroups - * - DeleteRecords - * - DeleteTopics - * - DeleteConsumerGroupOffsets - * - DescribeConfigs - * - CreateAcls - */ -rd_kafka_resp_err_t test_wait_topic_admin_result(rd_kafka_queue_t *q, - rd_kafka_event_type_t evtype, - rd_kafka_event_t **retevent, - int tmout) { - rd_kafka_event_t *rkev; - size_t i; - const rd_kafka_topic_result_t **terr = NULL; - size_t terr_cnt = 0; - const rd_kafka_ConfigResource_t **cres = NULL; - size_t cres_cnt = 0; - const rd_kafka_acl_result_t **aclres = NULL; - size_t aclres_cnt = 0; - int errcnt = 0; - rd_kafka_resp_err_t err; - const rd_kafka_group_result_t **gres = NULL; - size_t gres_cnt = 0; - const rd_kafka_ConsumerGroupDescription_t **gdescs = NULL; - size_t gdescs_cnt = 0; - const rd_kafka_error_t **glists_errors = NULL; - size_t glists_error_cnt = 0; - const rd_kafka_topic_partition_list_t *offsets = NULL; - - rkev = test_wait_admin_result(q, evtype, tmout); - - if ((err = rd_kafka_event_error(rkev))) { - TEST_WARN("%s failed: %s\n", rd_kafka_event_name(rkev), - rd_kafka_event_error_string(rkev)); - rd_kafka_event_destroy(rkev); - return err; - } - - if (evtype == RD_KAFKA_EVENT_CREATETOPICS_RESULT) { - const rd_kafka_CreateTopics_result_t *res; - if (!(res = rd_kafka_event_CreateTopics_result(rkev))) - TEST_FAIL("Expected a CreateTopics result, not %s", - rd_kafka_event_name(rkev)); - - terr = rd_kafka_CreateTopics_result_topics(res, &terr_cnt); - - } else if (evtype == RD_KAFKA_EVENT_DELETETOPICS_RESULT) { - const rd_kafka_DeleteTopics_result_t *res; - if (!(res = rd_kafka_event_DeleteTopics_result(rkev))) - TEST_FAIL("Expected a DeleteTopics result, not %s", - rd_kafka_event_name(rkev)); - - terr = rd_kafka_DeleteTopics_result_topics(res, &terr_cnt); - - } else if (evtype == RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT) { - const rd_kafka_CreatePartitions_result_t *res; - if (!(res = rd_kafka_event_CreatePartitions_result(rkev))) - TEST_FAIL("Expected a CreatePartitions result, not %s", - rd_kafka_event_name(rkev)); - - terr = rd_kafka_CreatePartitions_result_topics(res, &terr_cnt); - - } else if (evtype == RD_KAFKA_EVENT_DESCRIBECONFIGS_RESULT) { - const rd_kafka_DescribeConfigs_result_t *res; - - if (!(res = rd_kafka_event_DescribeConfigs_result(rkev))) - TEST_FAIL("Expected a DescribeConfigs result, not %s", - rd_kafka_event_name(rkev)); - - cres = - rd_kafka_DescribeConfigs_result_resources(res, &cres_cnt); - - } else if (evtype == RD_KAFKA_EVENT_ALTERCONFIGS_RESULT) { - const rd_kafka_AlterConfigs_result_t *res; - - if (!(res = rd_kafka_event_AlterConfigs_result(rkev))) - TEST_FAIL("Expected a AlterConfigs result, not %s", - rd_kafka_event_name(rkev)); - - cres = rd_kafka_AlterConfigs_result_resources(res, &cres_cnt); - - } else if (evtype == RD_KAFKA_EVENT_CREATEACLS_RESULT) { - const rd_kafka_CreateAcls_result_t *res; - - if (!(res = rd_kafka_event_CreateAcls_result(rkev))) - TEST_FAIL("Expected a CreateAcls result, not %s", - rd_kafka_event_name(rkev)); - - aclres = rd_kafka_CreateAcls_result_acls(res, &aclres_cnt); - } else if (evtype == RD_KAFKA_EVENT_LISTCONSUMERGROUPS_RESULT) { - const rd_kafka_ListConsumerGroups_result_t *res; - if (!(res = rd_kafka_event_ListConsumerGroups_result(rkev))) - TEST_FAIL( - "Expected a ListConsumerGroups result, not %s", - rd_kafka_event_name(rkev)); - - glists_errors = rd_kafka_ListConsumerGroups_result_errors( - res, &glists_error_cnt); - } else if (evtype == RD_KAFKA_EVENT_DESCRIBECONSUMERGROUPS_RESULT) { - const rd_kafka_DescribeConsumerGroups_result_t *res; - if (!(res = rd_kafka_event_DescribeConsumerGroups_result(rkev))) - TEST_FAIL( - "Expected a DescribeConsumerGroups result, not %s", - rd_kafka_event_name(rkev)); - - gdescs = rd_kafka_DescribeConsumerGroups_result_groups( - res, &gdescs_cnt); - } else if (evtype == RD_KAFKA_EVENT_DELETEGROUPS_RESULT) { - const rd_kafka_DeleteGroups_result_t *res; - if (!(res = rd_kafka_event_DeleteGroups_result(rkev))) - TEST_FAIL("Expected a DeleteGroups result, not %s", - rd_kafka_event_name(rkev)); - - gres = rd_kafka_DeleteGroups_result_groups(res, &gres_cnt); - - } else if (evtype == RD_KAFKA_EVENT_DELETERECORDS_RESULT) { - const rd_kafka_DeleteRecords_result_t *res; - if (!(res = rd_kafka_event_DeleteRecords_result(rkev))) - TEST_FAIL("Expected a DeleteRecords result, not %s", - rd_kafka_event_name(rkev)); - - offsets = rd_kafka_DeleteRecords_result_offsets(res); - - } else if (evtype == RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT) { - const rd_kafka_DeleteConsumerGroupOffsets_result_t *res; - if (!(res = rd_kafka_event_DeleteConsumerGroupOffsets_result( - rkev))) - TEST_FAIL( - "Expected a DeleteConsumerGroupOffsets " - "result, not %s", - rd_kafka_event_name(rkev)); - - gres = rd_kafka_DeleteConsumerGroupOffsets_result_groups( - rkev, &gres_cnt); - - } else { - TEST_FAIL("Bad evtype: %d", evtype); - RD_NOTREACHED(); - } - - /* Check topic errors */ - for (i = 0; i < terr_cnt; i++) { - if (rd_kafka_topic_result_error(terr[i])) { - TEST_WARN("..Topics result: %s: error: %s\n", - rd_kafka_topic_result_name(terr[i]), - rd_kafka_topic_result_error_string(terr[i])); - if (!(errcnt++)) - err = rd_kafka_topic_result_error(terr[i]); - } - } - - /* Check resource errors */ - for (i = 0; i < cres_cnt; i++) { - if (rd_kafka_ConfigResource_error(cres[i])) { - TEST_WARN( - "ConfigResource result: %d,%s: error: %s\n", - rd_kafka_ConfigResource_type(cres[i]), - rd_kafka_ConfigResource_name(cres[i]), - rd_kafka_ConfigResource_error_string(cres[i])); - if (!(errcnt++)) - err = rd_kafka_ConfigResource_error(cres[i]); - } - } - - /* Check ACL errors */ - for (i = 0; i < aclres_cnt; i++) { - const rd_kafka_error_t *error = - rd_kafka_acl_result_error(aclres[i]); - if (error) { - TEST_WARN("AclResult error: %s: %s\n", - rd_kafka_error_name(error), - rd_kafka_error_string(error)); - if (!(errcnt++)) - err = rd_kafka_error_code(error); - } - } - - /* Check list groups errors */ - for (i = 0; i < glists_error_cnt; i++) { - const rd_kafka_error_t *error = glists_errors[i]; - TEST_WARN("%s error: %s\n", rd_kafka_event_name(rkev), - rd_kafka_error_string(error)); - if (!(errcnt++)) - err = rd_kafka_error_code(error); - } - - /* Check describe groups errors */ - for (i = 0; i < gdescs_cnt; i++) { - const rd_kafka_error_t *error; - if ((error = - rd_kafka_ConsumerGroupDescription_error(gdescs[i]))) { - TEST_WARN("%s result: %s: error: %s\n", - rd_kafka_event_name(rkev), - rd_kafka_ConsumerGroupDescription_group_id( - gdescs[i]), - rd_kafka_error_string(error)); - if (!(errcnt++)) - err = rd_kafka_error_code(error); - } - } - - /* Check group errors */ - for (i = 0; i < gres_cnt; i++) { - const rd_kafka_topic_partition_list_t *parts; - - if (rd_kafka_group_result_error(gres[i])) { - - TEST_WARN("%s result: %s: error: %s\n", - rd_kafka_event_name(rkev), - rd_kafka_group_result_name(gres[i]), - rd_kafka_error_string( - rd_kafka_group_result_error(gres[i]))); - if (!(errcnt++)) - err = rd_kafka_error_code( - rd_kafka_group_result_error(gres[i])); - } - - parts = rd_kafka_group_result_partitions(gres[i]); - if (parts) { - int j; - for (j = 0; j < parts->cnt; i++) { - if (!parts->elems[j].err) - continue; - - TEST_WARN( - "%s result: %s: " - "%s [%" PRId32 "] error: %s\n", - rd_kafka_event_name(rkev), - rd_kafka_group_result_name(gres[i]), - parts->elems[j].topic, - parts->elems[j].partition, - rd_kafka_err2str(parts->elems[j].err)); - errcnt++; - } - } - } - - /* Check offset errors */ - for (i = 0; (offsets && i < (size_t)offsets->cnt); i++) { - if (offsets->elems[i].err) { - TEST_WARN("DeleteRecords result: %s [%d]: error: %s\n", - offsets->elems[i].topic, - offsets->elems[i].partition, - rd_kafka_err2str(offsets->elems[i].err)); - if (!(errcnt++)) - err = offsets->elems[i].err; - } - } - - if (!err && retevent) - *retevent = rkev; - else - rd_kafka_event_destroy(rkev); - - return err; -} - - - -/** - * @brief Topic Admin API helpers - * - * @param useq Makes the call async and posts the response in this queue. - * If NULL this call will be synchronous and return the error - * result. - * - * @remark Fails the current test on failure. - */ - -rd_kafka_resp_err_t test_CreateTopics_simple(rd_kafka_t *rk, - rd_kafka_queue_t *useq, - char **topics, - size_t topic_cnt, - int num_partitions, - void *opaque) { - rd_kafka_NewTopic_t **new_topics; - rd_kafka_AdminOptions_t *options; - rd_kafka_queue_t *q; - size_t i; - const int tmout = 30 * 1000; - rd_kafka_resp_err_t err; - - new_topics = malloc(sizeof(*new_topics) * topic_cnt); - - for (i = 0; i < topic_cnt; i++) { - char errstr[512]; - new_topics[i] = rd_kafka_NewTopic_new( - topics[i], num_partitions, 1, errstr, sizeof(errstr)); - TEST_ASSERT(new_topics[i], - "Failed to NewTopic(\"%s\", %d) #%" PRIusz ": %s", - topics[i], num_partitions, i, errstr); - } - - options = rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_CREATETOPICS); - rd_kafka_AdminOptions_set_opaque(options, opaque); - - if (!useq) { - char errstr[512]; - - err = rd_kafka_AdminOptions_set_request_timeout( - options, tmout, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_request_timeout: %s", errstr); - err = rd_kafka_AdminOptions_set_operation_timeout( - options, tmout - 5000, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_operation_timeout: %s", errstr); - - q = rd_kafka_queue_new(rk); - } else { - q = useq; - } - - TEST_SAY("Creating %" PRIusz " topics\n", topic_cnt); - - rd_kafka_CreateTopics(rk, new_topics, topic_cnt, options, q); - - rd_kafka_AdminOptions_destroy(options); - - rd_kafka_NewTopic_destroy_array(new_topics, topic_cnt); - free(new_topics); - - if (useq) - return RD_KAFKA_RESP_ERR_NO_ERROR; - - - err = test_wait_topic_admin_result( - q, RD_KAFKA_EVENT_CREATETOPICS_RESULT, NULL, tmout + 5000); - - rd_kafka_queue_destroy(q); - - if (err) - TEST_FAIL("Failed to create %d topic(s): %s", (int)topic_cnt, - rd_kafka_err2str(err)); - - return err; -} - - -rd_kafka_resp_err_t test_CreatePartitions_simple(rd_kafka_t *rk, - rd_kafka_queue_t *useq, - const char *topic, - size_t total_part_cnt, - void *opaque) { - rd_kafka_NewPartitions_t *newp[1]; - rd_kafka_AdminOptions_t *options; - rd_kafka_queue_t *q; - const int tmout = 30 * 1000; - rd_kafka_resp_err_t err; - char errstr[512]; - - newp[0] = rd_kafka_NewPartitions_new(topic, total_part_cnt, errstr, - sizeof(errstr)); - TEST_ASSERT(newp[0], "Failed to NewPartitions(\"%s\", %" PRIusz "): %s", - topic, total_part_cnt, errstr); - - options = - rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_CREATEPARTITIONS); - rd_kafka_AdminOptions_set_opaque(options, opaque); - - if (!useq) { - err = rd_kafka_AdminOptions_set_request_timeout( - options, tmout, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_request_timeout: %s", errstr); - err = rd_kafka_AdminOptions_set_operation_timeout( - options, tmout - 5000, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_operation_timeout: %s", errstr); - - q = rd_kafka_queue_new(rk); - } else { - q = useq; - } - - TEST_SAY("Creating (up to) %" PRIusz " partitions for topic \"%s\"\n", - total_part_cnt, topic); - - rd_kafka_CreatePartitions(rk, newp, 1, options, q); - - rd_kafka_AdminOptions_destroy(options); - - rd_kafka_NewPartitions_destroy(newp[0]); - - if (useq) - return RD_KAFKA_RESP_ERR_NO_ERROR; - - - err = test_wait_topic_admin_result( - q, RD_KAFKA_EVENT_CREATEPARTITIONS_RESULT, NULL, tmout + 5000); - - rd_kafka_queue_destroy(q); - - if (err) - TEST_FAIL("Failed to create partitions: %s", - rd_kafka_err2str(err)); - - return err; -} - - -rd_kafka_resp_err_t test_DeleteTopics_simple(rd_kafka_t *rk, - rd_kafka_queue_t *useq, - char **topics, - size_t topic_cnt, - void *opaque) { - rd_kafka_queue_t *q; - rd_kafka_DeleteTopic_t **del_topics; - rd_kafka_AdminOptions_t *options; - size_t i; - rd_kafka_resp_err_t err; - const int tmout = 30 * 1000; - - del_topics = malloc(sizeof(*del_topics) * topic_cnt); - - for (i = 0; i < topic_cnt; i++) { - del_topics[i] = rd_kafka_DeleteTopic_new(topics[i]); - TEST_ASSERT(del_topics[i]); - } - - options = rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_DELETETOPICS); - rd_kafka_AdminOptions_set_opaque(options, opaque); - - if (!useq) { - char errstr[512]; - - err = rd_kafka_AdminOptions_set_request_timeout( - options, tmout, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_request_timeout: %s", errstr); - err = rd_kafka_AdminOptions_set_operation_timeout( - options, tmout - 5000, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_operation_timeout: %s", errstr); - - q = rd_kafka_queue_new(rk); - } else { - q = useq; - } - - TEST_SAY("Deleting %" PRIusz " topics\n", topic_cnt); - - rd_kafka_DeleteTopics(rk, del_topics, topic_cnt, options, useq); - - rd_kafka_AdminOptions_destroy(options); - - rd_kafka_DeleteTopic_destroy_array(del_topics, topic_cnt); - - free(del_topics); - - if (useq) - return RD_KAFKA_RESP_ERR_NO_ERROR; - - err = test_wait_topic_admin_result( - q, RD_KAFKA_EVENT_DELETETOPICS_RESULT, NULL, tmout + 5000); - - rd_kafka_queue_destroy(q); - - if (err) - TEST_FAIL("Failed to delete topics: %s", rd_kafka_err2str(err)); - - return err; -} - -rd_kafka_resp_err_t test_DeleteGroups_simple(rd_kafka_t *rk, - rd_kafka_queue_t *useq, - char **groups, - size_t group_cnt, - void *opaque) { - rd_kafka_queue_t *q; - rd_kafka_DeleteGroup_t **del_groups; - rd_kafka_AdminOptions_t *options; - size_t i; - rd_kafka_resp_err_t err; - const int tmout = 30 * 1000; - - del_groups = malloc(sizeof(*del_groups) * group_cnt); - - for (i = 0; i < group_cnt; i++) { - del_groups[i] = rd_kafka_DeleteGroup_new(groups[i]); - TEST_ASSERT(del_groups[i]); - } - - options = rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_DELETEGROUPS); - rd_kafka_AdminOptions_set_opaque(options, opaque); - - if (!useq) { - char errstr[512]; - - err = rd_kafka_AdminOptions_set_request_timeout( - options, tmout, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_request_timeout: %s", errstr); - - q = rd_kafka_queue_new(rk); - } else { - q = useq; - } - - TEST_SAY("Deleting %" PRIusz " groups\n", group_cnt); - - rd_kafka_DeleteGroups(rk, del_groups, group_cnt, options, q); - - rd_kafka_AdminOptions_destroy(options); - - rd_kafka_DeleteGroup_destroy_array(del_groups, group_cnt); - free(del_groups); - - if (useq) - return RD_KAFKA_RESP_ERR_NO_ERROR; - - err = test_wait_topic_admin_result( - q, RD_KAFKA_EVENT_DELETEGROUPS_RESULT, NULL, tmout + 5000); - - rd_kafka_queue_destroy(q); - - if (err) - TEST_FAIL("Failed to delete groups: %s", rd_kafka_err2str(err)); - - return err; -} - -rd_kafka_resp_err_t -test_DeleteRecords_simple(rd_kafka_t *rk, - rd_kafka_queue_t *useq, - const rd_kafka_topic_partition_list_t *offsets, - void *opaque) { - rd_kafka_queue_t *q; - rd_kafka_AdminOptions_t *options; - rd_kafka_resp_err_t err; - rd_kafka_DeleteRecords_t *del_records = - rd_kafka_DeleteRecords_new(offsets); - const int tmout = 30 * 1000; - - options = - rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_DELETERECORDS); - rd_kafka_AdminOptions_set_opaque(options, opaque); - - if (!useq) { - char errstr[512]; - - err = rd_kafka_AdminOptions_set_request_timeout( - options, tmout, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_request_timeout: %s", errstr); - err = rd_kafka_AdminOptions_set_operation_timeout( - options, tmout - 5000, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_operation_timeout: %s", errstr); - - q = rd_kafka_queue_new(rk); - } else { - q = useq; - } - - TEST_SAY("Deleting offsets from %d partitions\n", offsets->cnt); - - rd_kafka_DeleteRecords(rk, &del_records, 1, options, q); - - rd_kafka_DeleteRecords_destroy(del_records); - - rd_kafka_AdminOptions_destroy(options); - - if (useq) - return RD_KAFKA_RESP_ERR_NO_ERROR; - - err = test_wait_topic_admin_result( - q, RD_KAFKA_EVENT_DELETERECORDS_RESULT, NULL, tmout + 5000); - - rd_kafka_queue_destroy(q); - - if (err) - TEST_FAIL("Failed to delete records: %s", - rd_kafka_err2str(err)); - - return err; -} - -rd_kafka_resp_err_t test_DeleteConsumerGroupOffsets_simple( - rd_kafka_t *rk, - rd_kafka_queue_t *useq, - const char *group_id, - const rd_kafka_topic_partition_list_t *offsets, - void *opaque) { - rd_kafka_queue_t *q; - rd_kafka_AdminOptions_t *options; - rd_kafka_resp_err_t err; - const int tmout = 30 * 1000; - rd_kafka_DeleteConsumerGroupOffsets_t *cgoffsets; - - options = rd_kafka_AdminOptions_new( - rk, RD_KAFKA_ADMIN_OP_DELETECONSUMERGROUPOFFSETS); - rd_kafka_AdminOptions_set_opaque(options, opaque); - - if (!useq) { - char errstr[512]; - - err = rd_kafka_AdminOptions_set_request_timeout( - options, tmout, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_request_timeout: %s", errstr); - err = rd_kafka_AdminOptions_set_operation_timeout( - options, tmout - 5000, errstr, sizeof(errstr)); - TEST_ASSERT(!err, "set_operation_timeout: %s", errstr); - - q = rd_kafka_queue_new(rk); - } else { - q = useq; - } - - if (offsets) { - TEST_SAY( - "Deleting committed offsets for group %s and " - "%d partitions\n", - group_id, offsets->cnt); - - cgoffsets = - rd_kafka_DeleteConsumerGroupOffsets_new(group_id, offsets); - } else { - TEST_SAY("Provoking invalid DeleteConsumerGroupOffsets call\n"); - cgoffsets = NULL; - } - - rd_kafka_DeleteConsumerGroupOffsets(rk, &cgoffsets, cgoffsets ? 1 : 0, - options, useq); - - if (cgoffsets) - rd_kafka_DeleteConsumerGroupOffsets_destroy(cgoffsets); - - rd_kafka_AdminOptions_destroy(options); - - if (useq) - return RD_KAFKA_RESP_ERR_NO_ERROR; - - err = test_wait_topic_admin_result( - q, RD_KAFKA_EVENT_DELETECONSUMERGROUPOFFSETS_RESULT, NULL, - tmout + 5000); - - rd_kafka_queue_destroy(q); - - if (err) - TEST_FAIL("Failed to delete committed offsets: %s", - rd_kafka_err2str(err)); - - return err; -} - -/** - * @brief Delta Alter configuration for the given resource, - * overwriting/setting the configs provided in \p configs. - * Existing configuration remains intact. - * - * @param configs 'const char *name, const char *value' tuples - * @param config_cnt is the number of tuples in \p configs - */ -rd_kafka_resp_err_t test_AlterConfigs_simple(rd_kafka_t *rk, - rd_kafka_ResourceType_t restype, - const char *resname, - const char **configs, - size_t config_cnt) { - rd_kafka_queue_t *q; - rd_kafka_ConfigResource_t *confres; - rd_kafka_event_t *rkev; - size_t i; - rd_kafka_resp_err_t err; - const rd_kafka_ConfigResource_t **results; - size_t result_cnt; - const rd_kafka_ConfigEntry_t **configents; - size_t configent_cnt; - - - q = rd_kafka_queue_new(rk); - - TEST_SAY("Getting configuration for %d %s\n", restype, resname); - - confres = rd_kafka_ConfigResource_new(restype, resname); - rd_kafka_DescribeConfigs(rk, &confres, 1, NULL, q); - - err = test_wait_topic_admin_result( - q, RD_KAFKA_EVENT_DESCRIBECONFIGS_RESULT, &rkev, 15 * 1000); - if (err) { - rd_kafka_queue_destroy(q); - rd_kafka_ConfigResource_destroy(confres); - return err; - } - - results = rd_kafka_DescribeConfigs_result_resources( - rd_kafka_event_DescribeConfigs_result(rkev), &result_cnt); - TEST_ASSERT(result_cnt == 1, - "expected 1 DescribeConfigs result, not %" PRIusz, - result_cnt); - - configents = - rd_kafka_ConfigResource_configs(results[0], &configent_cnt); - TEST_ASSERT(configent_cnt > 0, - "expected > 0 ConfigEntry:s, not %" PRIusz, configent_cnt); - - TEST_SAY("Altering configuration for %d %s\n", restype, resname); - - /* Apply all existing configuration entries to resource object that - * will later be passed to AlterConfigs. */ - for (i = 0; i < configent_cnt; i++) { - const char *entry_name = - rd_kafka_ConfigEntry_name(configents[i]); - - if (test_broker_version >= TEST_BRKVER(3, 2, 0, 0)) { - /* Skip entries that are overwritten to - * avoid duplicates, that cause an error since - * this broker version. */ - size_t j; - for (j = 0; j < config_cnt; j += 2) { - if (!strcmp(configs[j], entry_name)) { - break; - } - } - - if (j < config_cnt) - continue; - } - - err = rd_kafka_ConfigResource_set_config( - confres, entry_name, - rd_kafka_ConfigEntry_value(configents[i])); - TEST_ASSERT(!err, - "Failed to set read-back config %s=%s " - "on local resource object", - entry_name, - rd_kafka_ConfigEntry_value(configents[i])); - } - - rd_kafka_event_destroy(rkev); - - /* Then apply the configuration to change. */ - for (i = 0; i < config_cnt; i += 2) { - err = rd_kafka_ConfigResource_set_config(confres, configs[i], - configs[i + 1]); - TEST_ASSERT(!err, - "Failed to set config %s=%s on " - "local resource object", - configs[i], configs[i + 1]); - } - - rd_kafka_AlterConfigs(rk, &confres, 1, NULL, q); - - rd_kafka_ConfigResource_destroy(confres); - - err = test_wait_topic_admin_result( - q, RD_KAFKA_EVENT_ALTERCONFIGS_RESULT, NULL, 15 * 1000); - - rd_kafka_queue_destroy(q); - - return err; -} - -/** - * @brief Topic Admin API helpers - * - * @param useq Makes the call async and posts the response in this queue. - * If NULL this call will be synchronous and return the error - * result. - * - * @remark Fails the current test on failure. - */ - -rd_kafka_resp_err_t test_CreateAcls_simple(rd_kafka_t *rk, - rd_kafka_queue_t *useq, - rd_kafka_AclBinding_t **acls, - size_t acl_cnt, - void *opaque) { - rd_kafka_AdminOptions_t *options; - rd_kafka_queue_t *q; - rd_kafka_resp_err_t err; - const int tmout = 30 * 1000; - - options = rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_CREATEACLS); - rd_kafka_AdminOptions_set_opaque(options, opaque); - - if (!useq) { - q = rd_kafka_queue_new(rk); - } else { - q = useq; - } - - TEST_SAY("Creating %" PRIusz " acls\n", acl_cnt); - - rd_kafka_CreateAcls(rk, acls, acl_cnt, options, q); - - rd_kafka_AdminOptions_destroy(options); - - if (useq) - return RD_KAFKA_RESP_ERR_NO_ERROR; - - err = test_wait_topic_admin_result(q, RD_KAFKA_EVENT_CREATEACLS_RESULT, - NULL, tmout + 5000); - - rd_kafka_queue_destroy(q); - - if (err) - TEST_FAIL("Failed to create %d acl(s): %s", (int)acl_cnt, - rd_kafka_err2str(err)); - - return err; -} - -static void test_free_string_array(char **strs, size_t cnt) { - size_t i; - for (i = 0; i < cnt; i++) - free(strs[i]); - free(strs); -} - - -/** - * @return an array of all topics in the cluster matching our the - * rdkafka test prefix. - */ -static rd_kafka_resp_err_t -test_get_all_test_topics(rd_kafka_t *rk, char ***topicsp, size_t *topic_cntp) { - size_t test_topic_prefix_len = strlen(test_topic_prefix); - const rd_kafka_metadata_t *md; - char **topics = NULL; - size_t topic_cnt = 0; - int i; - rd_kafka_resp_err_t err; - - *topic_cntp = 0; - if (topicsp) - *topicsp = NULL; - - /* Retrieve list of topics */ - err = rd_kafka_metadata(rk, 1 /*all topics*/, NULL, &md, - tmout_multip(10000)); - if (err) { - TEST_WARN( - "%s: Failed to acquire metadata: %s: " - "not deleting any topics\n", - __FUNCTION__, rd_kafka_err2str(err)); - return err; - } - - if (md->topic_cnt == 0) { - TEST_WARN("%s: No topics in cluster\n", __FUNCTION__); - rd_kafka_metadata_destroy(md); - return RD_KAFKA_RESP_ERR_NO_ERROR; - } - - if (topicsp) - topics = malloc(sizeof(*topics) * md->topic_cnt); - - for (i = 0; i < md->topic_cnt; i++) { - if (strlen(md->topics[i].topic) >= test_topic_prefix_len && - !strncmp(md->topics[i].topic, test_topic_prefix, - test_topic_prefix_len)) { - if (topicsp) - topics[topic_cnt++] = - rd_strdup(md->topics[i].topic); - else - topic_cnt++; - } - } - - if (topic_cnt == 0) { - TEST_SAY( - "%s: No topics (out of %d) matching our " - "test prefix (%s)\n", - __FUNCTION__, md->topic_cnt, test_topic_prefix); - rd_kafka_metadata_destroy(md); - if (topics) - test_free_string_array(topics, topic_cnt); - return RD_KAFKA_RESP_ERR_NO_ERROR; - } - - rd_kafka_metadata_destroy(md); - - if (topicsp) - *topicsp = topics; - *topic_cntp = topic_cnt; - - return RD_KAFKA_RESP_ERR_NO_ERROR; -} - -/** - * @brief Delete all test topics using the Kafka Admin API. - */ -rd_kafka_resp_err_t test_delete_all_test_topics(int timeout_ms) { - rd_kafka_t *rk; - char **topics; - size_t topic_cnt = 0; - rd_kafka_resp_err_t err; - int i; - rd_kafka_AdminOptions_t *options; - rd_kafka_queue_t *q; - char errstr[256]; - int64_t abs_timeout = test_clock() + ((int64_t)timeout_ms * 1000); - - rk = test_create_producer(); - - err = test_get_all_test_topics(rk, &topics, &topic_cnt); - if (err) { - /* Error already reported by test_get_all_test_topics() */ - rd_kafka_destroy(rk); - return err; - } - - if (topic_cnt == 0) { - rd_kafka_destroy(rk); - return RD_KAFKA_RESP_ERR_NO_ERROR; - } - - q = rd_kafka_queue_get_main(rk); - - options = rd_kafka_AdminOptions_new(rk, RD_KAFKA_ADMIN_OP_DELETETOPICS); - if (rd_kafka_AdminOptions_set_operation_timeout(options, 2 * 60 * 1000, - errstr, sizeof(errstr))) - TEST_SAY(_C_YEL - "Failed to set DeleteTopics timeout: %s: " - "ignoring\n", - errstr); - - TEST_SAY(_C_MAG - "====> Deleting all test topics with <====" - "a timeout of 2 minutes\n"); - - test_DeleteTopics_simple(rk, q, topics, topic_cnt, options); - - rd_kafka_AdminOptions_destroy(options); - - while (1) { - rd_kafka_event_t *rkev; - const rd_kafka_DeleteTopics_result_t *res; - - rkev = rd_kafka_queue_poll(q, -1); - - res = rd_kafka_event_DeleteTopics_result(rkev); - if (!res) { - TEST_SAY("%s: Ignoring event: %s: %s\n", __FUNCTION__, - rd_kafka_event_name(rkev), - rd_kafka_event_error_string(rkev)); - rd_kafka_event_destroy(rkev); - continue; - } - - if (rd_kafka_event_error(rkev)) { - TEST_WARN("%s: DeleteTopics for %" PRIusz - " topics " - "failed: %s\n", - __FUNCTION__, topic_cnt, - rd_kafka_event_error_string(rkev)); - err = rd_kafka_event_error(rkev); - } else { - const rd_kafka_topic_result_t **terr; - size_t tcnt; - int okcnt = 0; - - terr = rd_kafka_DeleteTopics_result_topics(res, &tcnt); - - for (i = 0; i < (int)tcnt; i++) { - if (!rd_kafka_topic_result_error(terr[i])) { - okcnt++; - continue; - } - - TEST_WARN("%s: Failed to delete topic %s: %s\n", - __FUNCTION__, - rd_kafka_topic_result_name(terr[i]), - rd_kafka_topic_result_error_string( - terr[i])); - } - - TEST_SAY( - "%s: DeleteTopics " - "succeeded for %d/%" PRIusz " topics\n", - __FUNCTION__, okcnt, topic_cnt); - err = RD_KAFKA_RESP_ERR_NO_ERROR; - } - - rd_kafka_event_destroy(rkev); - break; - } - - rd_kafka_queue_destroy(q); - - test_free_string_array(topics, topic_cnt); - - /* Wait for topics to be fully deleted */ - while (1) { - err = test_get_all_test_topics(rk, NULL, &topic_cnt); - - if (!err && topic_cnt == 0) - break; - - if (abs_timeout < test_clock()) { - TEST_WARN( - "%s: Timed out waiting for " - "remaining %" PRIusz - " deleted topics " - "to disappear from cluster metadata\n", - __FUNCTION__, topic_cnt); - break; - } - - TEST_SAY("Waiting for remaining %" PRIusz - " delete topics " - "to disappear from cluster metadata\n", - topic_cnt); - - rd_sleep(1); - } - - rd_kafka_destroy(rk); - - return err; -} - - - -void test_fail0(const char *file, - int line, - const char *function, - int do_lock, - int fail_now, - const char *fmt, - ...) { - char buf[512]; - int is_thrd = 0; - size_t of; - va_list ap; - char *t; - char timestr[32]; - time_t tnow = time(NULL); - -#ifdef __MINGW32__ - strftime(timestr, sizeof(timestr), "%a %b %d %H:%M:%S %Y", - localtime(&tnow)); -#elif defined(_WIN32) - ctime_s(timestr, sizeof(timestr), &tnow); -#else - ctime_r(&tnow, timestr); -#endif - t = strchr(timestr, '\n'); - if (t) - *t = '\0'; - - of = rd_snprintf(buf, sizeof(buf), "%s%s%s():%i: ", test_curr->subtest, - *test_curr->subtest ? ": " : "", function, line); - rd_assert(of < sizeof(buf)); - - va_start(ap, fmt); - rd_vsnprintf(buf + of, sizeof(buf) - of, fmt, ap); - va_end(ap); - - /* Remove trailing newline */ - if ((t = strchr(buf, '\n')) && !*(t + 1)) - *t = '\0'; - - TEST_SAYL(0, "TEST FAILURE\n"); - fprintf(stderr, - "\033[31m### Test \"%s%s%s%s\" failed at %s:%i:%s() at %s: " - "###\n" - "%s\n", - test_curr->name, *test_curr->subtest ? " (" : "", - test_curr->subtest, *test_curr->subtest ? ")" : "", file, line, - function, timestr, buf + of); - if (do_lock) - TEST_LOCK(); - test_curr->state = TEST_FAILED; - test_curr->failcnt += 1; - test_curr->is_fatal_cb = NULL; - - if (!*test_curr->failstr) { - strncpy(test_curr->failstr, buf, sizeof(test_curr->failstr)); - test_curr->failstr[sizeof(test_curr->failstr) - 1] = '\0'; - } - if (fail_now && test_curr->mainfunc) { - tests_running_cnt--; - is_thrd = 1; - } - if (do_lock) - TEST_UNLOCK(); - if (!fail_now) - return; - if (test_assert_on_fail || !is_thrd) - assert(0); - else - thrd_exit(0); -} - - -/** - * @brief Destroy a mock cluster and its underlying rd_kafka_t handle - */ -void test_mock_cluster_destroy(rd_kafka_mock_cluster_t *mcluster) { - rd_kafka_t *rk = rd_kafka_mock_cluster_handle(mcluster); - rd_kafka_mock_cluster_destroy(mcluster); - rd_kafka_destroy(rk); -} - - - -/** - * @brief Create a standalone mock cluster that can be used by multiple - * rd_kafka_t instances. - */ -rd_kafka_mock_cluster_t *test_mock_cluster_new(int broker_cnt, - const char **bootstraps) { - rd_kafka_t *rk; - rd_kafka_conf_t *conf = rd_kafka_conf_new(); - rd_kafka_mock_cluster_t *mcluster; - char errstr[256]; - - test_conf_common_init(conf, 0); - - test_conf_set(conf, "client.id", "MOCK"); - - rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr)); - TEST_ASSERT(rk, "Failed to create mock cluster rd_kafka_t: %s", errstr); - - mcluster = rd_kafka_mock_cluster_new(rk, broker_cnt); - TEST_ASSERT(mcluster, "Failed to acquire mock cluster"); - - if (bootstraps) - *bootstraps = rd_kafka_mock_cluster_bootstraps(mcluster); - - return mcluster; -} - - - -/** - * @name Sub-tests - */ - - -/** - * @brief Start a sub-test. \p fmt is optional and allows additional - * sub-test info to be displayed, e.g., test parameters. - * - * @returns 0 if sub-test should not be run, else 1. - */ -int test_sub_start(const char *func, - int line, - int is_quick, - const char *fmt, - ...) { - - if (!is_quick && test_quick) - return 0; - - if (fmt && *fmt) { - va_list ap; - char buf[256]; - - va_start(ap, fmt); - rd_vsnprintf(buf, sizeof(buf), fmt, ap); - va_end(ap); - - rd_snprintf(test_curr->subtest, sizeof(test_curr->subtest), - "%s:%d: %s", func, line, buf); - } else { - rd_snprintf(test_curr->subtest, sizeof(test_curr->subtest), - "%s:%d", func, line); - } - - if (subtests_to_run && !strstr(test_curr->subtest, subtests_to_run)) { - *test_curr->subtest = '\0'; - return 0; - } - - test_curr->subtest_quick = is_quick; - - TIMING_START(&test_curr->subtest_duration, "SUBTEST"); - - TEST_SAY(_C_MAG "[ %s ]\n", test_curr->subtest); - - return 1; -} - - -/** - * @brief Reset the current subtest state. - */ -static void test_sub_reset(void) { - *test_curr->subtest = '\0'; - test_curr->is_fatal_cb = NULL; - test_curr->ignore_dr_err = rd_false; - test_curr->exp_dr_err = RD_KAFKA_RESP_ERR_NO_ERROR; - /* Don't check msg status by default */ - test_curr->exp_dr_status = (rd_kafka_msg_status_t)-1; - test_curr->dr_mv = NULL; -} - -/** - * @brief Sub-test has passed. - */ -void test_sub_pass(void) { - - TEST_ASSERT(*test_curr->subtest); - - TEST_SAYL(1, _C_GRN "[ %s: PASS (%.02fs) ]\n", test_curr->subtest, - (float)(TIMING_DURATION(&test_curr->subtest_duration) / - 1000000.0f)); - - if (test_curr->subtest_quick && test_quick && !test_on_ci && - TIMING_DURATION(&test_curr->subtest_duration) > 45 * 1000 * 1000) - TEST_WARN( - "Subtest %s marked as QUICK but took %.02fs to " - "finish: either fix the test or " - "remove the _QUICK identifier (limit is 45s)\n", - test_curr->subtest, - (float)(TIMING_DURATION(&test_curr->subtest_duration) / - 1000000.0f)); - - test_sub_reset(); -} - - -/** - * @brief Skip sub-test (must have been started with SUB_TEST*()). - */ -void test_sub_skip(const char *fmt, ...) { - va_list ap; - char buf[256]; - - TEST_ASSERT(*test_curr->subtest); - - va_start(ap, fmt); - rd_vsnprintf(buf, sizeof(buf), fmt, ap); - va_end(ap); - - TEST_SAYL(1, _C_YEL "[ %s: SKIP: %s ]\n", test_curr->subtest, buf); - - test_sub_reset(); -} |