...

So, This KIP try to raise one solution to improve it.

Public Interfaces

No public interface changed. Just change the inner implement of private method:

org.apache.kafka.clients.producer.KafkaProducer#doSend

The solution is that we can provide one method for metadata fetch into Producer.  When the application restarted/started, it can call it before mark the application is ready for handle requests. 

public Cluster getCluster(String topic, long maxBlockTimeMs) {
        Objects.requireNonNull(topic, "topic cannot be null");
        try {
            return waitOnMetadata(topic, null, time.milliseconds(), maxBlockTimeMs).cluster;
        } catch (InterruptedException e) {
            throw new InterruptException(e);
}

note: waitOnMetadata(topic, null, time.milliseconds(), maxBlockTimeMs) is the existed method with provide modifier.

Public Interfaces

add new interface with tiny refactor which reduce the duplicated code.

Cluster getCluster(String topic, long maxBlockTimeMs);Add two new configure items for producer.

Proposed Changes

The changes can refer to the example PR:   https://github.com/apache/kafka/pull/13320/files

By default, includeWaitTimeOnMetadataInMaxBlockTime is true, all of the behaviors are not changed.

When user set includeWaitTimeOnMetadataInMaxBlockTime to false, KafkaProducer#send will block maxWaitTimeMsOnMetadata for metadata's fetch and block max.block.ms for remaining operations.The core code had been listed in Motivation part.

Compatibility, Deprecation, and Migration Plan

...

• What impact (if any) will there be on existing users?  
  no impact on existed users.
• If we are changing behavior how will we phase out the older behavior?
  no changing older behavior.
• If we need special migration tools, describe them here.
  no.
• When will we remove the existing behavior?
  no need to remove.

Test Plan

Compare the first record's send time cost to see if any improvement happens.

Rejected Alternatives

maybe we can provide one dedicated method with more naming instead of "getCluster".