...

Specific modifications would be to make KafkaProducer#waitOnMetadata to be asynchronous when it must block. A client For uncached topics, the producer will maintain a queue of records for uncached topics will be maintained its outstanding records to ensure proper ordering of submission and callback invocation, where the records would flow back into the current execution logic when (in the accumulator and for callback invocations) once topic metadata is resolved. Proper care must be taken to handle batch sizing and the efforts to maintain the linger period for fetching metadata, and individual record timeout while queued.

To address (2), the producer currently maintains a staleness duration an expiry threshold for every topic, but it does not act upon this for metadata fetching, instead falls back to fetching information about all recently used topics. Further optimization could be done to only request metadata updates for topics whose staleness thresholds have been exceeded. A soft threshold could also be added such that best-effort fetching could be performed on a subset of the topics, so that metadata updates are staggered over time and performed in smaller batcheswhich is used to remove a topic from the working set at a future time (currently hard-coded to 5 minutes). While this works to reduce the size of the topic working set by evicting topics that are no longer in use, the producer will continue fetching metadata for these topics in every metadata request for the full expiry duration. This logic can be made more intelligent by managing the expiry from when the topic was last used, enabling the expiry duration to be reduced to improve cases where a lot of topics are touched intermittently. To control behavior, a configuration variable topic.expiry.ms should be added to the producer configuration.

Additionally, a second configuration variable topic.refresh.ms may be introduced to control the permitted staleness of a topic's metadata. Then, when topic metadata is fetched, only topics whose metadata is older than the refresh value need to be included in the request. Background, best-effort updates should be performed for topics with stale metadata, which would work towards reducing the number of topics included in a single topic metadata request.

Compatibility, Deprecation, and Migration Plan

...