deleteTopicRecords = async ({ topic, partitions }) => {
    if (!topic || typeof topic !== 'string') {
      throw new KafkaJSNonRetriableError(`Invalid topic "${topic}"`)
    }

    if (!partitions || partitions.length === 0) {
      throw new KafkaJSNonRetriableError(`Invalid partitions`)
    }

    const partitionsByBroker = cluster.findLeaderForPartitions(
      topic,
      partitions.map(p => p.partition)
    )

    const partitionsFound = values(partitionsByBroker).flat()
    const topicOffsets = await fetchTopicOffsets(topic)

    const leaderNotFoundErrors = []
    partitions.forEach(({ partition, offset }) => {
      // throw if no leader found for partition
      if (!partitionsFound.includes(partition)) {
        leaderNotFoundErrors.push({
          partition,
          offset,
          error: new KafkaJSBrokerNotFound('Could not find the leader for the partition', {
            retriable: false,
          }),
        })
        return
      }
      const { low } = topicOffsets.find(p => p.partition === partition) || {
        high: undefined,
        low: undefined,
      }
      // warn in case of offset below low watermark
      if (parseInt(offset) < parseInt(low) && parseInt(offset) !== -1) {
        logger.warn(
          'The requested offset is before the earliest offset maintained on the partition - no records will be deleted from this partition',
          {
            topic,
            partition,
            offset,
          }
        )
      }
    })

    if (leaderNotFoundErrors.length > 0) {
      throw new KafkaJSDeleteTopicRecordsError({ topic, partitions: leaderNotFoundErrors })
    }

    const seekEntriesByBroker = entries(partitionsByBroker).reduce(
      (obj, [nodeId, nodePartitions]) => {
        obj[nodeId] = {
          topic,
          partitions: partitions.filter(p => nodePartitions.includes(p.partition)),
        }
        return obj
      },
      {}
    )

    const retrier = createRetry(retry)
    return retrier(async bail => {
      try {
        const partitionErrors = []

        const brokerRequests = entries(seekEntriesByBroker).map(
          ([nodeId, { topic, partitions }]) => async () => {
            const broker = await cluster.findBroker({ nodeId })
            await broker.deleteRecords({ topics: [{ topic, partitions }] })
            // remove successful entry so it's ignored on retry
            delete seekEntriesByBroker[nodeId]
          }
        )

        await Promise.all(
          brokerRequests.map(request =>
            request().catch(e => {
              if (e.name === 'KafkaJSDeleteTopicRecordsError') {
                e.partitions.forEach(({ partition, offset, error }) => {
                  partitionErrors.push({
                    partition,
                    offset,
                    error,
                  })
                })
              } else {
                // then it's an unknown error, not from the broker response
                throw e
              }
            })
          )
        )

        if (partitionErrors.length > 0) {
          throw new KafkaJSDeleteTopicRecordsError({
            topic,
            partitions: partitionErrors,
          })
        }
      } catch (e) {
        if (
          e.retriable &&
          e.partitions.some(
            ({ error }) => staleMetadata(error) || error.name === 'KafkaJSMetadataNotLoaded'
          )
        ) {
          await cluster.refreshMetadata()
        }
        throw e
      }
    })
  }
Delete topic records up to the selected partition offsets

@param {string} topic
@param {Array<SeekEntry>} partitions
@return {Promise}

@typedef {Object} SeekEntry
@property {number} partition
@property {string} offset