#NEWLINE# def get_log_client( log_link ) : #NEWLINE# #INDENT# global _log_client_queue #NEWLINE# global MAX_HEAP_SIZE #NEWLINE# _log_client_lock . acquire ( ) #NEWLINE# try : #NEWLINE# #INDENT# components = urlparse . urlsplit ( new_client_tuple ) #NEWLINE# base_url = ( '%(scheme)s://%(netloc)s' % { #NEWLINE# #INDENT# 'scheme' : components [ 0 ] , #NEWLINE# 'netloc' : components [ 1 ] , #NEWLINE# #UNINDENT# } ) #NEWLINE# client_tuple = next ( ( tup for tup in _log_client_heap if ( tup [ 1 ] . base_url == base_url ) ) , None ) #NEWLINE# if ( client_tuple is None ) : #NEWLINE# #INDENT# client = HttpClient ( base_url , logger = LOG ) #NEWLINE# yarn_cluster = cluster . get_cluster_conf_for_job_submission ( ) #NEWLINE# if yarn_cluster . SECURITY_ENABLED . get ( ) : #NEWLINE# #INDENT# client . set_kerberos_auth ( ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# _log_client_heap . remove ( client_tuple ) #NEWLINE# client = client_tuple [ 1 ] #NEWLINE# #UNINDENT# new_client_tuple = ( time . time ( ) , client ) #NEWLINE# if ( len ( _log_client_heap ) >= MAX_HEAP_SIZE ) : #NEWLINE# #INDENT# heapq . heapreplace ( _log_client_heap , new_client_tuple ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# heapq . heappush ( _log_client_heap , new_client_tuple ) #NEWLINE# #UNINDENT# return client #NEWLINE# #UNINDENT# finally : #NEWLINE# #INDENT# _log_client_lock . release ( )