1616log = logging .getLogger (__name__ )
1717
1818CLUSTER = None
19+ CLUSTER_CONF = None
20+ CLUSTER_DEFAULTS = {
21+ "ZOOKEEPER_PORT_OFFSET" : 20000 ,
22+ "ZOOKEEPER_CLUSTER_SIZE" : 3 ,
23+ "ZOOKEEPER_OBSERVER_START_ID" : - 1 ,
24+ }
1925
2026
2127def get_global_cluster ():
22- global CLUSTER
23- if CLUSTER is None :
24- ZK_HOME = os .environ .get ("ZOOKEEPER_PATH" )
25- ZK_CLASSPATH = os .environ .get ("ZOOKEEPER_CLASSPATH" )
26- ZK_PORT_OFFSET = int (os .environ .get ("ZOOKEEPER_PORT_OFFSET" , 20000 ))
27- ZK_CLUSTER_SIZE = int (os .environ .get ("ZOOKEEPER_CLUSTER_SIZE" , 3 ))
28- ZK_VERSION = os .environ .get ("ZOOKEEPER_VERSION" )
29- if '-' in ZK_VERSION :
30- # Ignore pre-release markers like -alpha
31- ZK_VERSION = ZK_VERSION .split ('-' )[0 ]
32- ZK_VERSION = tuple ([int (n ) for n in ZK_VERSION .split ('.' )])
33-
34- ZK_OBSERVER_START_ID = int (
35- os .environ .get ("ZOOKEEPER_OBSERVER_START_ID" , - 1 ))
36-
37- assert ZK_HOME or ZK_CLASSPATH or ZK_VERSION , (
38- "Either ZOOKEEPER_PATH or ZOOKEEPER_CLASSPATH or "
39- "ZOOKEEPER_VERSION environment variable must be defined.\n "
40- "For deb package installations this is /usr/share/java" )
41-
42- if ZK_VERSION >= (3 , 5 ):
43- additional_configuration_entries = [
44- "4lw.commands.whitelist=*" ,
45- "reconfigEnabled=true"
46- ]
47- # If defines, this sets the superuser password to "test"
48- additional_java_system_properties = [
49- "-Dzookeeper.DigestAuthenticationProvider.superDigest="
50- "super:D/InIHSb7yEEbrWz8b9l71RjZJU="
51- ]
28+ global CLUSTER , CLUSTER_CONF
29+ cluster_conf = {
30+ k : os .environ .get (k , CLUSTER_DEFAULTS .get (k ))
31+ for k in ["ZOOKEEPER_PATH" ,
32+ "ZOOKEEPER_CLASSPATH" ,
33+ "ZOOKEEPER_PORT_OFFSET" ,
34+ "ZOOKEEPER_CLUSTER_SIZE" ,
35+ "ZOOKEEPER_VERSION" ,
36+ "ZOOKEEPER_OBSERVER_START_ID" ,
37+ "ZOOKEEPER_JAAS_AUTH" ]
38+ }
39+ if CLUSTER is not None :
40+ if CLUSTER_CONF == cluster_conf :
41+ return CLUSTER
5242 else :
53- additional_configuration_entries = []
54- additional_java_system_properties = []
55- CLUSTER = ZookeeperCluster (
56- install_path = ZK_HOME ,
57- classpath = ZK_CLASSPATH ,
58- port_offset = ZK_PORT_OFFSET ,
59- size = ZK_CLUSTER_SIZE ,
60- observer_start_id = ZK_OBSERVER_START_ID ,
61- configuration_entries = additional_configuration_entries ,
62- java_system_properties = additional_java_system_properties
63- )
64- atexit .register (lambda cluster : cluster .terminate (), CLUSTER )
43+ log .info ('Config change detected. Reconfiguring cluster...' )
44+ CLUSTER .terminate ()
45+ CLUSTER = None
46+ # Create a new cluster
47+ ZK_HOME = cluster_conf .get ("ZOOKEEPER_PATH" )
48+ ZK_CLASSPATH = cluster_conf .get ("ZOOKEEPER_CLASSPATH" )
49+ ZK_PORT_OFFSET = int (cluster_conf .get ("ZOOKEEPER_PORT_OFFSET" ))
50+ ZK_CLUSTER_SIZE = int (cluster_conf .get ("ZOOKEEPER_CLUSTER_SIZE" ))
51+ ZK_VERSION = cluster_conf .get ("ZOOKEEPER_VERSION" )
52+ if '-' in ZK_VERSION :
53+ # Ignore pre-release markers like -alpha
54+ ZK_VERSION = ZK_VERSION .split ('-' )[0 ]
55+ ZK_VERSION = tuple ([int (n ) for n in ZK_VERSION .split ('.' )])
56+ ZK_OBSERVER_START_ID = int (cluster_conf .get ("ZOOKEEPER_OBSERVER_START_ID" ))
57+
58+ assert ZK_HOME or ZK_CLASSPATH or ZK_VERSION , (
59+ "Either ZOOKEEPER_PATH or ZOOKEEPER_CLASSPATH or "
60+ "ZOOKEEPER_VERSION environment variable must be defined.\n "
61+ "For deb package installations this is /usr/share/java" )
62+
63+ if ZK_VERSION >= (3 , 5 ):
64+ additional_configuration_entries = [
65+ "4lw.commands.whitelist=*" ,
66+ "reconfigEnabled=true"
67+ ]
68+ # If defined, this sets the superuser password to "test"
69+ additional_java_system_properties = [
70+ "-Dzookeeper.DigestAuthenticationProvider.superDigest="
71+ "super:D/InIHSb7yEEbrWz8b9l71RjZJU="
72+ ]
73+ else :
74+ additional_configuration_entries = []
75+ additional_java_system_properties = []
76+ ZOOKEEPER_JAAS_AUTH = cluster_conf .get ("ZOOKEEPER_JAAS_AUTH" )
77+ if ZOOKEEPER_JAAS_AUTH == "digest" :
78+ jaas_config = """
79+ Server {
80+ org.apache.zookeeper.server.auth.DigestLoginModule required
81+ user_super="super_secret"
82+ user_jaasuser="jaas_password";
83+ };"""
84+ elif ZOOKEEPER_JAAS_AUTH == "gssapi" :
85+ # Configure Zookeeper to use our test KDC.
86+ additional_java_system_properties += [
87+ "-Djava.security.krb5.conf=%s" % os .path .expandvars (
88+ "${KRB5_CONFIG}"
89+ ),
90+ "-Dsun.security.krb5.debug=true" ,
91+ ]
92+ jaas_config = """
93+ Server {
94+ com.sun.security.auth.module.Krb5LoginModule required
95+ debug=true
96+ useKeyTab=true
97+ keyTab="%s"
98+ storeKey=true
99+ useTicketCache=false
100+ principal="zookeeper/[email protected] "; 101+ };""" % os .path .expandvars ("${KRB5_TEST_ENV}/server.keytab" )
102+ else :
103+ jaas_config = None
104+
105+ CLUSTER = ZookeeperCluster (
106+ install_path = ZK_HOME ,
107+ classpath = ZK_CLASSPATH ,
108+ port_offset = ZK_PORT_OFFSET ,
109+ size = ZK_CLUSTER_SIZE ,
110+ observer_start_id = ZK_OBSERVER_START_ID ,
111+ configuration_entries = additional_configuration_entries ,
112+ java_system_properties = additional_java_system_properties ,
113+ jaas_config = jaas_config
114+ )
115+ CLUSTER_CONF = cluster_conf
116+ atexit .register (lambda cluster : cluster .terminate (), CLUSTER )
65117 return CLUSTER
66118
67119
@@ -90,6 +142,7 @@ def test_something_else(self):
90142 something_that_needs_zk_servers(self.servers)
91143
92144 """
145+ DEFAULT_CLIENT_TIMEOUT = 15
93146
94147 def __init__ (self , * args , ** kw ):
95148 super (KazooTestHarness , self ).__init__ (* args , ** kw )
@@ -109,8 +162,10 @@ def _get_nonchroot_client(self):
109162 self ._clients .append (c )
110163 return c
111164
112- def _get_client (self , ** kwargs ):
113- c = KazooClient (self .hosts , ** kwargs )
165+ def _get_client (self , ** client_options ):
166+ if 'timeout' not in client_options :
167+ client_options ['timeout' ] = self .DEFAULT_CLIENT_TIMEOUT
168+ c = KazooClient (self .hosts , ** client_options )
114169 self ._clients .append (c )
115170 return c
116171
@@ -139,7 +194,7 @@ def setup_zookeeper(self, **client_options):
139194 namespace = "/kazootests" + uuid .uuid4 ().hex
140195 self .hosts = self .servers + namespace
141196 if 'timeout' not in client_options :
142- client_options ['timeout' ] = 0.8
197+ client_options ['timeout' ] = self . DEFAULT_CLIENT_TIMEOUT
143198 self .client = self ._get_client (** client_options )
144199 self .client .start ()
145200 self .client .ensure_path ("/" )
0 commit comments