Metadata propagation should also work for existing topics

[marcin-krystianc]

Fixes: #4904

When a topic is created, all brokers do not immediately know about its existence. That leads to a race condition, where from a producer's perspective topic may seem to be created and deleted immediately afterward.
Thus we should wait for the topic propagation time before deleting the topic from the cache.

Test results (no more unknow topic or partition errors):

TEST 20241118220526 (bare, scenario default) SUMMARY
#==================================================================#
| <MAIN>                                   |     PASSED | 1039.781s |
| 0000_unittests                           |     PASSED |   4.709s |
| 0001_multiobj                            |     PASSED |   1.211s |
| 0002_unkpart                             |     PASSED |   2.047s |
| 0003_msgmaxsize                          |     PASSED |   1.135s |
| 0004_conf                                |     PASSED |   0.012s |
| 0005_order                               |     PASSED |   1.099s |
| 0006_symbols                             |     PASSED |   0.000s |
| 0007_autotopic                           |     PASSED |   1.062s |
| 0008_reqacks                             |     PASSED |   1.071s |
| 0009_mock_cluster                        |     PASSED |   0.121s |
| 0011_produce_batch                       |     PASSED |  10.181s |
| 0012_produce_consume                     |     PASSED |   2.717s |
| 0013_null_msgs                           |     PASSED |   2.081s |
| 0014_reconsume_191                       |     PASSED |   7.871s |
| 0015_offsets_seek                        |     PASSED |  15.776s |
| 0016_client_swname                       |     PASSED |   0.018s |
| 0017_compression                         |     PASSED |   8.167s |
| 0018_cgrp_term                           |     PASSED |  39.007s |
| 0019_list_groups                         |     PASSED |   9.067s |
| 0020_destroy_hang                        |     PASSED |  17.570s |
| 0021_rkt_destroy                         |     PASSED |   1.936s |
| 0022_consume_batch                       |     PASSED |   4.157s |
| 0022_consume_batch_local                 |     PASSED |   1.001s |
| 0025_timers                              |     PASSED |   6.004s |
| 0026_consume_pause                       |     PASSED |  80.618s |
| 0028_long_topicnames                     |    SKIPPED |   0.000s | Filtered due to negative test flags
| 0029_assign_offset                       |     PASSED |   8.031s |
| 0030_offset_commit                       |     PASSED |  53.731s |
| 0031_get_offsets                         |     PASSED |   1.731s |
| 0031_get_offsets_mock                    |     PASSED |   1.047s |
| 0033_regex_subscribe                     |     PASSED |  88.442s |
| 0033_regex_subscribe_local               |     PASSED |   0.023s |
| 0034_offset_reset                        |     PASSED |   3.049s |
| 0034_offset_reset_mock                   |     PASSED |  12.747s |
| 0035_api_version                         |     PASSED |   0.005s |
| 0036_partial_fetch                       |     PASSED |   1.555s |
| 0037_destroy_hang_local                  |     PASSED |   1.002s |
| 0038_performance                         |     PASSED |   9.717s |
| 0039_event_dr                            |     PASSED |   0.040s |
| 0039_event_log                           |     PASSED |   0.001s |
| 0039_event                               |     PASSED |   0.001s |
| 0040_io_event                            |     PASSED |   6.071s |
| 0041_fetch_max_bytes                     |     PASSED |   5.916s |
| 0042_many_topics                         |     PASSED |  35.980s |
| 0043_no_connection                       |     PASSED |   1.001s |
| 0044_partition_cnt                       |     PASSED |  28.417s |
| 0045_subscribe_update                    |     PASSED |  25.347s |
| 0045_subscribe_update_mock               |     PASSED | 141.147s |
| 0045_subscribe_update_racks_mock         |     PASSED | 242.348s |
| 0046_rkt_cache                           |     PASSED |   0.001s |
| 0047_partial_buf_tmout                   |    SKIPPED |   0.000s | Filtered due to negative test flags
| 0048_partitioner                         |     PASSED |   2.082s |
| 0049_consume_conn_close                  |     PASSED |  22.550s |
| 0050_subscribe_adds                      |     PASSED |  33.193s |
| 0051_assign_adds                         |     PASSED |  12.124s |
| 0052_msg_timestamps                      |     PASSED |  57.860s |
| 0053_stats_timing                        |     PASSED |   1.203s |
| 0053_stats                               |    SKIPPED |   0.000s | RapidJSON >=1.1.0 not available
| 0054_offset_time                         |     PASSED |   2.805s |
| 0055_producer_latency                    |     PASSED |  39.635s |
| 0056_balanced_group_mt                   |     PASSED |   5.419s |
| 0057_invalid_topic                       |     PASSED |   1.884s |
| 0058_log                                 |     PASSED |   2.002s |
| 0059_bsearch                             |     PASSED |  11.034s |
| 0060_op_prio                             |     PASSED |   5.244s |
| 0061_consumer_lag                        |     PASSED |  23.027s |
| 0062_stats_event                         |     PASSED |   0.501s |
| 0063_clusterid                           |     PASSED |   5.344s |
| 0064_interceptors                        |     PASSED |   4.881s |
| 0065_yield                               |     PASSED |   3.345s |
| 0066_plugins                             |     PASSED |   0.004s |
| 0067_empty_topic                         |     PASSED |   4.115s |
| 0068_produce_timeout                     |     PASSED |  25.015s |
| 0069_consumer_add_parts                  |     PASSED |  20.097s |
| 0070_null_empty                          |     PASSED |   3.044s |
| 0072_headers_ut                          |     PASSED |   1.000s |
| 0073_headers                             |     PASSED |   1.524s |
| 0074_producev                            |     PASSED |   0.001s |
| 0075_retry                               |     PASSED |  15.014s |
| 0076_produce_retry                       |     PASSED |  52.018s |
| 0076_produce_retry_mock                  |     PASSED |   0.398s |
| 0077_compaction                          |     PASSED | 153.971s |
| 0078_c_from_cpp                          |     PASSED |   0.001s |
| 0079_fork                                |    SKIPPED |   0.000s | Filtered due to negative test flags
| 0080_admin_ut                            |     PASSED |  22.647s |
| 0081_admin                               |     FAILED | 809.388s | do_test_CreateTopics:75: 0081_admin#producer-322 CreateTopics with temp queue, op timeout 0, op_timeout 0, validate_only 0: do_test_CreateTopics():233: Expected INVALID_CONFIG, not 0: NO_ERROR
| 0082_fetch_max_bytes                     |     FAILED | 803.910s | test_error_cb():687: 0082_fetch_max_bytes#producer-325 rdkafka error: Local: Broker transport failure: localhost:57207/4: Receive failed: Socket operation on non-socket (after 4825ms in state APIVERSION_QUERY)
| 0083_cb_event                            |     PASSED |   8.048s |
| 0084_destroy_flags_local                 |     PASSED |   3.020s |
| 0084_destroy_flags                       |     PASSED |  18.744s |
| 0085_headers                             |     PASSED |   1.539s |
| 0086_purge_local                         |     PASSED |   0.001s |
| 0086_purge_remote                        |     PASSED |   6.021s |
| 0088_produce_metadata_timeout            |     PASSED |  21.006s |
| 0089_max_poll_interval                   |     PASSED | 106.451s |
| 0090_idempotence                         |     PASSED |   3.732s |
| 0091_max_poll_interval_timeout           |     PASSED |  65.322s |
| 0092_mixed_msgver                        |     PASSED |   4.281s |
| 0093_holb_consumer                       |     PASSED |  30.084s |
| 0094_idempotence_msg_timeout             |     PASSED |  60.922s |
| 0095_all_brokers_down                    |     PASSED |   2.003s |
| 0097_ssl_verify                          |    SKIPPED |   0.000s | Test requires SSL_* env-vars set up by trivup
| 0097_ssl_verify_local                    |     PASSED |   0.001s |
| 0098_consumer_txn                        |    SKIPPED |   0.000s | RapidJSON >=1.1.0 not available
| 0099_commit_metadata                     |     PASSED |   0.173s |
| 0100_thread_interceptors                 |     PASSED |   0.001s |
| 0101_fetch_from_follower                 |    SKIPPED |   0.000s | RapidJSON >=1.1.0 not available
| 0102_static_group_rebalance              |     PASSED |  50.858s |
| 0103_transactions_local                  |     PASSED |   7.002s |
| 0103_transactions                        |     PASSED |  88.624s |
| 0104_fetch_from_follower_mock            |     PASSED |  43.623s |
| 0105_transactions_mock                   |     PASSED | 372.187s |
| 0106_cgrp_sess_timeout                   |     PASSED |  57.314s |
| 0109_auto_create_topics                  |     PASSED |  34.209s |
| 0110_batch_size                          |    SKIPPED |   0.000s | RapidJSON >=1.1.0 not available
| 0112_assign_unknown_part                 |     PASSED |   3.054s |
| 0113_cooperative_rebalance_local         |     PASSED |  46.619s |
| 0113_cooperative_rebalance               |     PASSED | 618.050s |
| 0114_sticky_partitioning                 |     PASSED |  18.108s |
| 0115_producer_auth                       |     FAILED | 607.440s | ():77: Delivery report: Expected Broker: Topic authorization failed but got Success
| 0116_kafkaconsumer_close                 |     PASSED |  65.044s |
| 0117_mock_errors                         |     PASSED |  22.239s |
| 0118_commit_rebalance                    |     PASSED |  14.005s |
| 0119_consumer_auth                       |     PASSED |  22.135s |
| 0120_asymmetric_subscription             |     PASSED |  12.021s |
| 0121_clusterid                           |     PASSED |   5.003s |
| 0122_buffer_cleaning_after_rebalance     |     PASSED |  46.145s |
| 0123_connections_max_idle                |     PASSED |  16.002s |
| 0124_openssl_invalid_engine              |     PASSED |   0.017s |
| 0125_immediate_flush                     |     PASSED |  13.155s |
| 0125_immediate_flush_mock                |     PASSED |   0.046s |
| 0126_oauthbearer_oidc                    |    SKIPPED |   0.000s | Apache Kafka cluster not configured for SASL
| 0127_fetch_queue_backoff                 |     PASSED |   7.719s |
| 0128_sasl_callback_queue                 |     PASSED |  10.002s |
| 0129_fetch_aborted_msgs                  |     PASSED |  14.469s |
| 0130_store_offsets                       |     PASSED |   2.212s |
| 0131_connect_timeout                     |     PASSED |   8.001s |
| 0132_strategy_ordering                   |     PASSED |  16.098s |
| 0133_ssl_keys                            |     PASSED |   0.033s |
| 0134_ssl_provider                        |     PASSED |   0.002s |
| 0135_sasl_credentials                    |    SKIPPED |   0.000s | Test requires SASL PLAIN or SASL SCRAM
| 0136_resolve_cb                          |     PASSED |   1.029s |
| 0137_barrier_batch_consume               |     PASSED |  59.198s |
| 0138_admin_mock                          |     PASSED |   0.095s |
| 0139_offset_validation_mock              |     PASSED |  27.597s |
| 0140_commit_metadata                     |     PASSED |   0.040s |
| 0142_reauthentication                    |    SKIPPED |   0.000s | Test requires SASL_PLAINTEXT or SASL_SSL, got plaintext
| 0143_exponential_backoff_mock            |     PASSED |  83.016s |
| 0144_idempotence_mock                    |     PASSED |   9.846s |
| 0145_pause_resume_mock                   |     PASSED |  18.338s |
| 0146_metadata_mock                       |     PASSED |  12.890s |
| 0150_telemetry_mock                      |     PASSED | 153.462s |
#==================================================================#

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
✅ marcin-krystianc
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[adamreeve]

Hi @emasab, can you please take a look at this PR, or is there anybody else maintaining this library who can take a look?