QuorumPeerConfig { public void parseProperties(Properties zkProp) throws IOException, ConfigException { int clientPort = 0; int secureClientPort = 0; String clientPortAddress = null; String secureClientPortAddress = null; VerifyingFileFactory vff = new VerifyingFileFactory.Builder(LOG).warnForRelativePath().build(); for (Entry<Object, Object> entry : zkProp.entrySet()) { String key = entry.getKey().toString().trim(); String value = entry.getValue().toString().trim(); if (key.equals("dataDir")) { dataDir = vff.create(value); } else if (key.equals("dataLogDir")) { dataLogDir = vff.create(value); } else if (key.equals("clientPort")) { clientPort = Integer.parseInt(value); } else if (key.equals("localSessionsEnabled")) { localSessionsEnabled = Boolean.parseBoolean(value); } else if (key.equals("localSessionsUpgradingEnabled")) { localSessionsUpgradingEnabled = Boolean.parseBoolean(value); } else if (key.equals("clientPortAddress")) { clientPortAddress = value.trim(); } else if (key.equals("secureClientPort")) { secureClientPort = Integer.parseInt(value); } else if (key.equals("secureClientPortAddress")){ secureClientPortAddress = value.trim(); } else if (key.equals("tickTime")) { tickTime = Integer.parseInt(value); } else if (key.equals("maxClientCnxns")) { maxClientCnxns = Integer.parseInt(value); } else if (key.equals("minSessionTimeout")) { minSessionTimeout = Integer.parseInt(value); } else if (key.equals("maxSessionTimeout")) { maxSessionTimeout = Integer.parseInt(value); } else if (key.equals("initLimit")) { initLimit = Integer.parseInt(value); } else if (key.equals("syncLimit")) { syncLimit = Integer.parseInt(value); } else if (key.equals("electionAlg")) { electionAlg = Integer.parseInt(value); } else if (key.equals("quorumListenOnAllIPs")) { quorumListenOnAllIPs = Boolean.parseBoolean(value); } else if (key.equals("peerType")) { if (value.toLowerCase().equals("observer")) { peerType = LearnerType.OBSERVER; } else if (value.toLowerCase().equals("participant")) { peerType = LearnerType.PARTICIPANT; } else { throw new ConfigException("Unrecognised peertype: " + value); } } else if (key.equals( "syncEnabled" )) { syncEnabled = Boolean.parseBoolean(value); } else if (key.equals("dynamicConfigFile")){ dynamicConfigFileStr = value; } else if (key.equals("autopurge.snapRetainCount")) { snapRetainCount = Integer.parseInt(value); } else if (key.equals("autopurge.purgeInterval")) { purgeInterval = Integer.parseInt(value); } else if (key.equals("standaloneEnabled")) { if (value.toLowerCase().equals("true")) { setStandaloneEnabled(true); } else if (value.toLowerCase().equals("false")) { setStandaloneEnabled(false); } else { throw new ConfigException("Invalid option " + value + " for standalone mode. Choose 'true' or 'false.'"); } } else if (key.equals("reconfigEnabled")) { if (value.toLowerCase().equals("true")) { setReconfigEnabled(true); } else if (value.toLowerCase().equals("false")) { setReconfigEnabled(false); } else { throw new ConfigException("Invalid option " + value + " for reconfigEnabled flag. Choose 'true' or 'false.'"); } } else if (key.equals("sslQuorum")){ sslQuorum = Boolean.parseBoolean(value); } else if (key.equals("portUnification")){ shouldUsePortUnification = Boolean.parseBoolean(value); } else if (key.equals("sslQuorumReloadCertFiles")) { sslQuorumReloadCertFiles = Boolean.parseBoolean(value); } else if ((key.startsWith("server.") || key.startsWith("group") || key.startsWith("weight")) && zkProp.containsKey("dynamicConfigFile")) { throw new ConfigException("parameter: " + key + " must be in a separate dynamic config file"); } else if (key.equals(QuorumAuth.QUORUM_SASL_AUTH_ENABLED)) { quorumEnableSasl = Boolean.parseBoolean(value); } else if (key.equals(QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED)) { quorumServerRequireSasl = Boolean.parseBoolean(value); } else if (key.equals(QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED)) { quorumLearnerRequireSasl = Boolean.parseBoolean(value); } else if (key.equals(QuorumAuth.QUORUM_LEARNER_SASL_LOGIN_CONTEXT)) { quorumLearnerLoginContext = value; } else if (key.equals(QuorumAuth.QUORUM_SERVER_SASL_LOGIN_CONTEXT)) { quorumServerLoginContext = value; } else if (key.equals(QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL)) { quorumServicePrincipal = value; } else if (key.equals("quorum.cnxn.threads.size")) { quorumCnxnThreadsSize = Integer.parseInt(value); } else { System.setProperty("zookeeper." + key, value); } } if (!quorumEnableSasl && quorumServerRequireSasl) { throw new IllegalArgumentException( QuorumAuth.QUORUM_SASL_AUTH_ENABLED + " is disabled, so cannot enable " + QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED); } if (!quorumEnableSasl && quorumLearnerRequireSasl) { throw new IllegalArgumentException( QuorumAuth.QUORUM_SASL_AUTH_ENABLED + " is disabled, so cannot enable " + QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED); } if (!quorumLearnerRequireSasl && quorumServerRequireSasl) { throw new IllegalArgumentException( QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED + " is disabled, so cannot enable " + QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED); } if (snapRetainCount < MIN_SNAP_RETAIN_COUNT) { LOG.warn("Invalid autopurge.snapRetainCount: " + snapRetainCount + ". Defaulting to " + MIN_SNAP_RETAIN_COUNT); snapRetainCount = MIN_SNAP_RETAIN_COUNT; } if (dataDir == null) { throw new IllegalArgumentException("dataDir is not set"); } if (dataLogDir == null) { dataLogDir = dataDir; } if (clientPort == 0) { LOG.info("clientPort is not set"); if (clientPortAddress != null) { throw new IllegalArgumentException("clientPortAddress is set but clientPort is not set"); } } else if (clientPortAddress != null) { this.clientPortAddress = new InetSocketAddress( InetAddress.getByName(clientPortAddress), clientPort); LOG.info("clientPortAddress is {}", this.clientPortAddress.toString()); } else { this.clientPortAddress = new InetSocketAddress(clientPort); LOG.info("clientPortAddress is {}", this.clientPortAddress.toString()); } if (secureClientPort == 0) { LOG.info("secureClientPort is not set"); if (secureClientPortAddress != null) { throw new IllegalArgumentException("secureClientPortAddress is set but secureClientPort is not set"); } } else if (secureClientPortAddress != null) { this.secureClientPortAddress = new InetSocketAddress( InetAddress.getByName(secureClientPortAddress), secureClientPort); LOG.info("secureClientPortAddress is {}", this.secureClientPortAddress.toString()); } else { this.secureClientPortAddress = new InetSocketAddress(secureClientPort); LOG.info("secureClientPortAddress is {}", this.secureClientPortAddress.toString()); } if (this.secureClientPortAddress != null) { configureSSLAuth(); } if (tickTime == 0) { throw new IllegalArgumentException("tickTime is not set"); } minSessionTimeout = minSessionTimeout == -1 ? tickTime * 2 : minSessionTimeout; maxSessionTimeout = maxSessionTimeout == -1 ? tickTime * 20 : maxSessionTimeout; if (minSessionTimeout > maxSessionTimeout) { throw new IllegalArgumentException( "minSessionTimeout must not be larger than maxSessionTimeout"); } if (dynamicConfigFileStr == null) { setupQuorumPeerConfig(zkProp, true); if (isDistributed() && isReconfigEnabled()) { backupOldConfig(); } } }  void parse(String path); static String getVersionFromFilename(String filename); void parseProperties(Properties zkProp); static void writeDynamicConfig(final String dynamicConfigFilename,
                                          final QuorumVerifier qv,
                                          final boolean needKeepVersion); static void editStaticConfig(final String configFileStr,
                                        final String dynamicFileStr,
                                        final boolean eraseClientPortAddress); static void deleteFile(String filename); static QuorumVerifier parseDynamicConfig(Properties dynamicConfigProp, int eAlg, boolean warnings,
	   boolean configBackwardCompatibilityMode); void checkValidity(); InetSocketAddress getClientPortAddress(); InetSocketAddress getSecureClientPortAddress(); File getDataDir(); File getDataLogDir(); int getTickTime(); int getMaxClientCnxns(); int getMinSessionTimeout(); int getMaxSessionTimeout(); boolean areLocalSessionsEnabled(); boolean isLocalSessionsUpgradingEnabled(); boolean isSslQuorum(); boolean shouldUsePortUnification(); int getInitLimit(); int getSyncLimit(); int getElectionAlg(); int getElectionPort(); int getSnapRetainCount(); int getPurgeInterval(); boolean getSyncEnabled(); QuorumVerifier getQuorumVerifier(); QuorumVerifier getLastSeenQuorumVerifier(); Map<Long,QuorumServer> getServers(); long getServerId(); boolean isDistributed(); LearnerType getPeerType(); String getConfigFilename(); Boolean getQuorumListenOnAllIPs(); static boolean isStandaloneEnabled(); static void setStandaloneEnabled(boolean enabled); static boolean isReconfigEnabled(); static void setReconfigEnabled(boolean enabled); static final String nextDynamicConfigFileSuffix; }
@Test public void testErrorMessageWhensecureClientPortNotSetButsecureClientPortAddressSet() throws IOException, ConfigException { QuorumPeerConfig quorumPeerConfig = new QuorumPeerConfig(); try { Properties zkProp = getDefaultZKProperties(); zkProp.setProperty("secureClientPortAddress", "localhost"); quorumPeerConfig.parseProperties(zkProp); fail("IllegalArgumentException is expected"); } catch (IllegalArgumentException e) { String expectedMessage = "secureClientPortAddress is set but secureClientPort is not set"; assertEquals(expectedMessage, e.getMessage()); } }
@Test public void testErrorMessageWhenclientPortNotSetButclientPortAddressSet() throws IOException, ConfigException { QuorumPeerConfig quorumPeerConfig = new QuorumPeerConfig(); try { Properties zkProp = getDefaultZKProperties(); zkProp.setProperty("clientPortAddress", "localhost"); quorumPeerConfig.parseProperties(zkProp); fail("IllegalArgumentException is expected"); } catch (IllegalArgumentException e) { String expectedMessage = "clientPortAddress is set but clientPort is not set"; assertEquals(expectedMessage, e.getMessage()); } }
@Test public void testConfigureSSLAuthGetsConfiguredIfSecurePortConfigured() throws IOException, ConfigException { String sslAuthProp = "zookeeper.authProvider.x509"; QuorumPeerConfig quorumPeerConfig = new QuorumPeerConfig(); Properties zkProp = getDefaultZKProperties(); zkProp.setProperty("secureClientPort", "12345"); quorumPeerConfig.parseProperties(zkProp); String expected = "org.apache.zookeeper.server.auth.X509AuthenticationProvider"; String result = System.getProperty(sslAuthProp); assertEquals(expected, result); }
@Test public void testCustomSSLAuth() throws IOException { try (ClientX509Util x509Util = new ClientX509Util()) { System.setProperty(x509Util.getSslAuthProviderProperty(), "y509"); QuorumPeerConfig quorumPeerConfig = new QuorumPeerConfig(); try { Properties zkProp = getDefaultZKProperties(); zkProp.setProperty("secureClientPort", "12345"); quorumPeerConfig.parseProperties(zkProp); fail("ConfigException is expected"); } catch (ConfigException e) { assertNotNull(e.getMessage()); } } }
@Test(expected = ConfigException.class) public void testSamePortConfiguredForClientAndElection() throws IOException, ConfigException { QuorumPeerConfig quorumPeerConfig = new QuorumPeerConfig(); Properties zkProp = getDefaultZKProperties(); zkProp.setProperty("server.1", "localhost:2888:2888"); quorumPeerConfig.parseProperties(zkProp); }