Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java (Java)

Code before:

        long remainingWaitMs = maxWaitMs;
        long elapsed = 0;
        // Issue metadata requests until we have metadata for the topic and the requested partition,
        // or until maxWaitTimeMs is exceeded. This is necessary in case the metadata
        // is stale and the number of partitions for this topic has increased in the meantime.
        long nowNanos = time.nanoseconds();
        do {
            if (partition != null) {
                log.trace("Requesting metadata update for partition {} of topic {}.", partition, topic);
            } else {
                log.trace("Requesting metadata update for topic {}.", topic);
            }
            metadata.add(topic, nowMs + elapsed);
            int version = metadata.requestUpdateForTopic(topic);
            sender.wakeup();
            try {
                metadata.awaitUpdate(version, remainingWaitMs);
            } catch (TimeoutException ex) {
                // Rethrow with original maxWaitMs to prevent logging exception with remainingWaitMs
                final String errorMessage = getErrorMessage(partitionsCount, topic, partition, maxWaitMs);
                if (metadata.getError(topic) != null) {
                    throw new TimeoutException(errorMessage, metadata.getError(topic).exception());
                }
                if (ex.getCause() != null)
                    throw new TimeoutException(errorMessage, ex.getCause());
                throw new TimeoutException(errorMessage, new KafkaException(METADATA_TIMEOUT_MSG));
            }
            cluster = metadata.fetch();
            elapsed = time.milliseconds() - nowMs;
            if (elapsed >= maxWaitMs) {
                final String errorMessage = getErrorMessage(partitionsCount, topic, partition, maxWaitMs);
                if (metadata.getError(topic) != null && metadata.getError(topic).exception() instanceof RetriableException) {
                    throw new TimeoutException(errorMessage, metadata.getError(topic).exception());
                }
                throw new TimeoutException(errorMessage, new KafkaException(METADATA_TIMEOUT_MSG));
            }
            metadata.maybeThrowExceptionForTopic(topic);
            remainingWaitMs = maxWaitMs - elapsed;
            partitionsCount = cluster.partitionCountForTopic(topic);
        } while (partitionsCount == null || (partition != null && partition >= partitionsCount));

        producerMetrics.recordMetadataWait(time.nanoseconds() - nowNanos);

        return new ClusterAndWaitTime(cluster, elapsed);
    }

    private String getErrorMessage(Integer partitionsCount, String topic, Integer partition, long maxWaitMs) {
        return partitionsCount == null ?
            String.format("Topic %s not present in metadata after %d ms.",
                topic, maxWaitMs) :

Review comment (refactor):
Would it be better to refactor it like this:
```
Errors error = metadata.getError(topic);
if (error != null) {
    cause = error.exception();
} else if (ex.getCause() != null) {
    cause = ex.getCause();
} else {
    cause = new KafkaException(METADATA_TIMEOUT_MSG);
}

throw new TimeoutException(errorMessage, cause);
```
Also, since `ex.getCause() != null` can never be true based on the current `awaitUpdate` implementation, do we need to check for it?


Code after:
        // Return cached metadata if we have it, and if the record's partition is either undefined
        // or within the known partition range
        if (partitionsCount != null && (partition == null || partition < partitionsCount))
            return new ClusterAndWaitTime(cluster, 0);

        long remainingWaitMs = maxWaitMs;
        long elapsed = 0;
        // Issue metadata requests until we have metadata for the topic and the requested partition,
        // or until maxWaitTimeMs is exceeded. This is necessary in case the metadata
        // is stale and the number of partitions for this topic has increased in the meantime.
        long nowNanos = time.nanoseconds();
        do {
            if (partition != null) {
                log.trace("Requesting metadata update for partition {} of topic {}.", partition, topic);
            } else {
                log.trace("Requesting metadata update for topic {}.", topic);
            }
            metadata.add(topic, nowMs + elapsed);
            int version = metadata.requestUpdateForTopic(topic);
            sender.wakeup();
            try {
                metadata.awaitUpdate(version, remainingWaitMs);
            } catch (TimeoutException ex) {
                // Rethrow with original maxWaitMs to prevent logging exception with remainingWaitMs
                final String errorMessage = getErrorMessage(partitionsCount, topic, partition, maxWaitMs);
                if (metadata.getError(topic) != null) {
                    throw new TimeoutException(errorMessage, metadata.getError(topic).exception());
                }
                throw new TimeoutException(errorMessage);
            }
            cluster = metadata.fetch();
            elapsed = time.milliseconds() - nowMs;
            if (elapsed >= maxWaitMs) {
                final String errorMessage = getErrorMessage(partitionsCount, topic, partition, maxWaitMs);
                if (metadata.getError(topic) != null && metadata.getError(topic).exception() instanceof RetriableException) {
                    throw new TimeoutException(errorMessage, metadata.getError(topic).exception());
                }
                throw new TimeoutException(errorMessage);
            }
            metadata.maybeThrowExceptionForTopic(topic);
            remainingWaitMs = maxWaitMs - elapsed;
            partitionsCount = cluster.partitionCountForTopic(topic);
        } while (partitionsCount == null || (partition != null && partition >= partitionsCount));

        producerMetrics.recordMetadataWait(time.nanoseconds() - nowNanos);

        return new ClusterAndWaitTime(cluster, elapsed);
    }

    private String getErrorMessage(Integer partitionsCount, String topic, Integer partition, long maxWaitMs) {
        return partitionsCount == null ?