const createRetry = require('../retry')
const flatten = require('../utils/flatten')
const groupMessagesPerPartition = require('./groupMessagesPerPartition')
const createTopicData = require('./createTopicData')
const responseSerializer = require('./responseSerializer')

const { keys } = Object
const TOTAL_INDIVIDUAL_ATTEMPTS = 5
const staleMetadata = e =>
  ['UNKNOWN_TOPIC_OR_PARTITION', 'LEADER_NOT_AVAILABLE', 'NOT_LEADER_FOR_PARTITION'].includes(
    e.type
  )

module.exports = ({ logger, cluster, partitioner }) => {
  const retrier = createRetry({ retries: TOTAL_INDIVIDUAL_ATTEMPTS })

  return async ({ topic, messages, acks, timeout, compression }) => {
    await cluster.addTargetTopic(topic)
    const partitionMetadata = cluster.findTopicPartitionMetadata(topic)
    const messagesPerPartition = groupMessagesPerPartition({
      topic,
      partitionMetadata,
      messages,
      partitioner,
    })

    const partitions = keys(messagesPerPartition)
    const partitionsPerLeader = cluster.findLeaderForPartitions(topic, partitions)
    const leaders = keys(partitionsPerLeader)
    const responsePerBroker = new Map()

    for (let nodeId of leaders) {
      const broker = await cluster.findBroker({ nodeId })
      responsePerBroker.set(broker, null)
    }

    const produce = responsePerBroker => {
      const brokers = Array.from(responsePerBroker.keys())
      const brokersWithoutResponse = brokers.filter(broker => !responsePerBroker.get(broker))

      return brokersWithoutResponse.map(async broker => {
        const partitions = partitionsPerLeader[broker.nodeId]
        const topicData = createTopicData({ topic, partitions, messagesPerPartition })

        const response = await broker.produce({ acks, timeout, compression, topicData })
        responsePerBroker.set(broker, responseSerializer(response))
      })
    }

    const makeRequests = async (bail, retryCount, retryTime) => {
      try {
        await Promise.all(produce(responsePerBroker))
        const responses = Array.from(responsePerBroker.values())
        return flatten(responses)
      } catch (e) {
        if (staleMetadata(e)) {
          await cluster.refreshMetadata()
        }

        throw e
      }
    }

    return retrier(makeRequests).catch(e => {
      throw e.originalError || e
    })
  }
}

@@ -691,32 +691,139 @@
 getTopic(topic)%0A
+    const responsePerBroker = new Map()%0A%0A    const createProducerRequests = async responsePerBroker =%3E %7B%0A  
     const partit
@@ -878,24 +878,26 @@
 (topic)%0A    
+  
 const messag
@@ -951,15 +951,19 @@
     
+ 
+ 
 topic,%0A
+  
     
@@ -989,16 +989,18 @@
 ,%0A      
+ 
+ 
 messages
@@ -1001,16 +1001,18 @@
 ssages,%0A
+  
       pa
@@ -1026,20 +1026,24 @@
 er,%0A    
+  
 %7D)%0A%0A
+  
     cons
@@ -1084,24 +1084,26 @@
 tition)%0A    
+  
 const partit
@@ -1169,24 +1169,26 @@
 itions)%0A    
+  
 const leader
@@ -1221,49 +1221,11 @@
 er)%0A
+%0A
   
-  const responsePerBroker = new Map()%0A%0A
     
@@ -1250,24 +1250,26 @@
  leaders) %7B%0A
+  
       const 
@@ -1316,24 +1316,31 @@
 Id %7D)%0A      
+  if (!
 responsePerB
@@ -1349,52 +1349,32 @@
 ker.
-set
+has
 (broker
-, null)
+)) %7B
 %0A    
-%7D%0A%0A
     
-const produce =
+ 
  res
@@ -1379,37 +1379,69 @@
 esponsePerBroker
- =%3E %7B
+.set(broker, null)%0A        %7D%0A      %7D%0A
 %0A      const bro
@@ -2050,52 +2050,107 @@
     
-await Promise.all(produce(responsePerBroker)
+const requests = await createProducerRequests(responsePerBroker)%0A        await Promise.all(requests
 )%0A  

Use refreshed metadata when retrying individual nodes