<commit_before>const Connection = require('../network/connection')
const { KafkaJSNonRetriableError, KafkaJSConnectionError } = require('../errors')
const shuffle = require('../utils/shuffle')

const validateBrokers = brokers => {
  if (!brokers || brokers.length === 0) {
    throw new KafkaJSNonRetriableError(`Failed to connect: expected brokers array and got nothing`)
  }
}

module.exports = ({
  socketFactory,
  brokers,
  ssl,
  sasl,
  clientId,
  requestTimeout,
  enforceRequestTimeout,
  connectionTimeout,
  maxInFlightRequests,
  retry,
  logger,
  instrumentationEmitter = null,
}) => {
  let getNext

  // dynamic list of brokers
  if (typeof brokers === 'function') {
    getNext = async () => {
      try {
        const discovered = await brokers()
        const [host, port] = shuffle(discovered.brokers)[0].split(':')

        return {
          host,
          port,
          sasl: discovered.sasl,
        }
      } catch (e) {
        throw new KafkaJSConnectionError('dynamic brokers function crashed, retrying...')
      }
    }

    // static list of seed brokers
  } else {
    validateBrokers(brokers)

    const shuffledBrokers = shuffle(brokers)
    const size = brokers.length
    let index = 0

    getNext = () => {
      // Always rotate the seed broker
      const [seedHost, seedPort] = shuffledBrokers[index++ % size].split(':')

      return {
        host: seedHost,
        port: Number(seedPort),
      }
    }
  }

  return {
    build: async () => {
      const broker = await getNext()

      return new Connection({
        host: broker.host,
        port: broker.port,
        ssl,
        sasl: broker.sasl || sasl,
        clientId,
        socketFactory,
        connectionTimeout,
        requestTimeout,
        enforceRequestTimeout,
        maxInFlightRequests,
        instrumentationEmitter,
        retry,
        logger,
      })
    },
    assign: ({ host, port, rack }) => {
      return new Connection({
        host,
        port,
        rack,
        ssl,
        sasl,
        clientId,
        socketFactory,
        connectionTimeout,
        requestTimeout,
        enforceRequestTimeout,
        maxInFlightRequests,
        instrumentationEmitter,
        retry,
        logger,
      })
    },
  }
}
<commit_msg>Update connectionBuilder.js<commit_after>const Connection = require('../network/connection')
const { KafkaJSNonRetriableError, KafkaJSConnectionError } = require('../errors')
const shuffle = require('../utils/shuffle')

const validateBrokers = brokers => {
  if (!brokers || brokers.length === 0) {
    throw new KafkaJSNonRetriableError(`Failed to connect: expected brokers array and got nothing`)
  }
}

module.exports = ({
  socketFactory,
  brokers,
  ssl,
  sasl,
  clientId,
  requestTimeout,
  enforceRequestTimeout,
  connectionTimeout,
  maxInFlightRequests,
  retry,
  logger,
  instrumentationEmitter = null,
}) => {
  let getNext

  // dynamic list of brokers
  if (typeof brokers === 'function') {
    getNext = async () => {
      try {
        const discovered = await brokers()
        const [seedHost, seedPort] = shuffle(discovered.brokers)[0].split(':')

        return {
          host: seedHost,
          port: Number(seedPort),
          sasl: discovered.sasl,
        }
      } catch (e) {
        throw new KafkaJSConnectionError('dynamic brokers function crashed, retrying...')
      }
    }

    // static list of seed brokers
  } else {
    validateBrokers(brokers)

    const shuffledBrokers = shuffle(brokers)
    const size = brokers.length
    let index = 0

    getNext = () => {
      // Always rotate the seed broker
      const [seedHost, seedPort] = shuffledBrokers[index++ % size].split(':')

      return {
        host: seedHost,
        port: Number(seedPort),
        sasl,
      }
    }
  }

  return {
    build: async () => {
      const broker = await getNext()

      return new Connection({
        host: broker.host,
        port: broker.port,
        sasl: broker.sasl,
        ssl,
        clientId,
        socketFactory,
        connectionTimeout,
        requestTimeout,
        enforceRequestTimeout,
        maxInFlightRequests,
        instrumentationEmitter,
        retry,
        logger,
      })
    },
    assign: ({ host, port, rack }) => {
      return new Connection({
        host,
        port,
        rack,
        ssl,
        sasl,
        clientId,
        socketFactory,
        connectionTimeout,
        requestTimeout,
        enforceRequestTimeout,
        maxInFlightRequests,
        instrumentationEmitter,
        retry,
        logger,
      })
    },
  }
}

{'fraction_non_alphanumeric': 0.10683391003460208, 'fraction_numerical': 0.0012975778546712802, 'mean_word_length': 2.1123822341857337, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3930417', 'n_tokens_mistral': 1363, 'n_tokens_neox': 1340, 'n_words': 317}