def get_transaction_coordinator(transactional_id:)
      @logger.debug "Getting transaction coordinator for `#{transactional_id}`"

      refresh_metadata_if_necessary!

      if transactional_id.nil?
        # Get a random_broker
        @logger.debug "Transaction ID is not available. Choose a random broker."
        return random_broker
      else
        get_coordinator(Kafka::Protocol::COORDINATOR_TYPE_TRANSACTION, transactional_id)
      end
    end
def get_transaction_coordinator(transactional_id:)
      @logger.debug "Getting transaction coordinator for `#{transactional_id}`"

      refresh_metadata_if_necessary!

      if transactional_id.nil?
        # Get a random_broker
        @logger.debug "Transaction ID is not available. Choose a random broker."
        return random_broker
      else
        get_coordinator(Kafka::Protocol::COORDINATOR_TYPE_TRANSACTION, transactional_id)
      end
    end
Finds the broker acting as the coordinator of the given transaction.

 @param transactional_id: [String]
 @return [Broker] the broker that's currently coordinator.