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[bot]]

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