Generalize PulsarConfigurationLoader

pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfiguration.java

@@ -22,13 +22,15 @@
 
 import com.google.common.collect.Sets;
 import com.yahoo.pulsar.client.impl.auth.AuthenticationDisabled;
+import com.yahoo.pulsar.common.configuration.FieldContext;
+import com.yahoo.pulsar.common.configuration.PulsarConfiguration;
 
 /**
  *
  * Pulsar service configuration object.
  *
  */
-public class ServiceConfiguration {
+public class ServiceConfiguration implements PulsarConfiguration{
 
     /***** --- pulsar configuration --- ****/
     // Zookeeper quorum connection string

pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/FieldContext.java → pulsar-broker-common/src/main/java/com/yahoo/pulsar/common/configuration/FieldContext.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.yahoo.pulsar.broker;
+package com.yahoo.pulsar.common.configuration;
 
 import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;

pulsar-broker-common/src/main/java/com/yahoo/pulsar/common/configuration/PulsarConfiguration.java

@@ -0,0 +1,25 @@
+/**
+ * Copyright 2016 Yahoo Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.yahoo.pulsar.common.configuration;
+
+import java.util.Properties;
+
+public interface PulsarConfiguration {
+
+    public Properties getProperties();
+
+    public void setProperties(Properties properties);
+}

pulsar-broker-common/src/main/java/com/yahoo/pulsar/broker/ServiceConfigurationLoader.java → pulsar-broker-common/src/main/java/com/yahoo/pulsar/common/configuration/PulsarConfigurationLoader.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.yahoo.pulsar.broker;
+package com.yahoo.pulsar.common.configuration;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.yahoo.pulsar.common.util.FieldParser.update;
@@ -30,22 +30,23 @@
  * 
  *
  */
-public class ServiceConfigurationLoader {
+public class PulsarConfigurationLoader {
 
     /**
-     * Creates ServiceConfiguration and loads it with populated attribute values loaded from provided property file.
+     * Creates PulsarConfiguration and loads it with populated attribute values loaded from provided property file.
      * 
      * @param configFile
      * @throws IOException
      * @throws IllegalArgumentException
      */
-    public static ServiceConfiguration create(String configFile) throws IOException, IllegalArgumentException {
+    public static <T extends PulsarConfiguration> T create(String configFile,
+            Class<? extends PulsarConfiguration> clazz) throws IOException, IllegalArgumentException {
         checkNotNull(configFile);
-        return create(new FileInputStream(configFile));
+        return create(new FileInputStream(configFile), clazz);
     }
 
     /**
-     * Creates ServiceConfiguration and loads it with populated attribute values loaded from provided inputstream
+     * Creates PulsarConfiguration and loads it with populated attribute values loaded from provided inputstream
      * property file.
      * 
      * @param inStream
@@ -54,12 +55,13 @@ public static ServiceConfiguration create(String configFile) throws IOException,
      * @throws IllegalArgumentException
      *             if the input stream contains incorrect value type
      */
-    public static ServiceConfiguration create(InputStream inStream) throws IOException, IllegalArgumentException {
+    public static <T extends PulsarConfiguration> T create(InputStream inStream,
+            Class<? extends PulsarConfiguration> clazz) throws IOException, IllegalArgumentException {
         try {
             checkNotNull(inStream);
             Properties properties = new Properties();
             properties.load(inStream);
-            return (create(properties));
+            return (create(properties, clazz));
         } finally {
             if (inStream != null) {
                 inStream.close();
@@ -68,11 +70,17 @@ public static ServiceConfiguration create(InputStream inStream) throws IOExcepti
     }
 
     @SuppressWarnings({ "rawtypes", "unchecked" })
-    public static ServiceConfiguration create(Properties properties) throws IOException, IllegalArgumentException {
+    protected static <T extends PulsarConfiguration> T create(Properties properties,
+            Class<? extends PulsarConfiguration> clazz) throws IOException, IllegalArgumentException {
         checkNotNull(properties);
-        ServiceConfiguration configuration = new ServiceConfiguration();
-        configuration.setProperties(properties);
-        update((Map) properties, configuration);
+        T configuration = null;
+        try {
+            configuration = (T) clazz.newInstance();
+            configuration.setProperties(properties);
+            update((Map) properties, configuration);
+        } catch (InstantiationException | IllegalAccessException e) {
+            throw new IllegalArgumentException("Failed to instantiate " + clazz.getName(), e);
+        }
         return configuration;
     }
 

pulsar-broker/src/test/java/com/yahoo/pulsar/common/naming/ServiceConfigurationLoaderTest.java → pulsar-broker-common/src/test/java/com/yahoo/pulsar/common/configuration/PulsarConfigurationLoaderTest.java

@@ -13,50 +13,83 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.yahoo.pulsar.common.naming;
+package com.yahoo.pulsar.common.configuration;
 
-import static com.yahoo.pulsar.broker.ServiceConfigurationLoader.isComplete;
+import static com.yahoo.pulsar.common.configuration.PulsarConfigurationLoader.isComplete;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
 import java.io.InputStream;
+import java.io.OutputStreamWriter;
+import java.io.PrintWriter;
 import java.util.Properties;
 
 import org.testng.annotations.Test;
 
-import com.yahoo.pulsar.broker.FieldContext;
 import com.yahoo.pulsar.broker.ServiceConfiguration;
-import com.yahoo.pulsar.broker.ServiceConfigurationLoader;
 
-public class ServiceConfigurationLoaderTest {
+public class PulsarConfigurationLoaderTest {
 
     @Test
-    public void testServiceConfiguraitonLoadingStream() throws Exception {
-        final String fileName = "configurations/pulsar_broker_test.conf"; // test-resource file
-        InputStream stream = this.getClass().getClassLoader().getResourceAsStream(fileName);
-        final ServiceConfiguration serviceConfig = ServiceConfigurationLoader.create(stream);
+    public void testPulsarConfiguraitonLoadingStream() throws Exception {
+        File testConfigFile = new File("tmp." + System.currentTimeMillis() + ".properties");
+        if (testConfigFile.exists()) {
+            testConfigFile.delete();
+        }
+        final String zkServer = "z1.example.com,z2.example.com,z3.example.com";
+        PrintWriter printWriter = new PrintWriter(new OutputStreamWriter(new FileOutputStream(testConfigFile)));
+        printWriter.println("zookeeperServers=" + zkServer);
+        printWriter.println("globalZookeeperServers=gz1.example.com,gz2.example.com,gz3.example.com/foo");
+        printWriter.println("brokerDeleteInactiveTopicsEnabled=true");
+        printWriter.println("statusFilePath=/tmp/status.html");
+        printWriter.println("managedLedgerDefaultEnsembleSize=1");
+        printWriter.println("backlogQuotaDefaultLimitGB=18");
+        printWriter.println("clusterName=usc");
+        printWriter.println("brokerClientAuthenticationPlugin=test.xyz.client.auth.plugin");
+        printWriter.println("brokerClientAuthenticationParameters=role:my-role");
+        printWriter.println("superUserRoles=appid1,appid2");
+        printWriter.println("brokerServicePort=7777");
+        printWriter.println("managedLedgerDefaultMarkDeleteRateLimit=5.0");
+        printWriter.close();
+        testConfigFile.deleteOnExit();
+        InputStream stream = new FileInputStream(testConfigFile);
+        final ServiceConfiguration serviceConfig = PulsarConfigurationLoader.create(stream, ServiceConfiguration.class);
         assertNotNull(serviceConfig);
+        assertEquals(serviceConfig.getZookeeperServers(), zkServer);
+        assertEquals(serviceConfig.isBrokerDeleteInactiveTopicsEnabled(), true);
+        assertEquals(serviceConfig.getBacklogQuotaDefaultLimitGB(), 18);
+        assertEquals(serviceConfig.getClusterName(), "usc");
+        assertEquals(serviceConfig.getBrokerClientAuthenticationParameters(), "role:my-role");
+        assertEquals(serviceConfig.getBrokerServicePort(), 7777);
     }
 
     @Test
-    public void testServiceConfiguraitonLoadingProp() throws Exception {
+    public void testPulsarConfiguraitonLoadingProp() throws Exception {
         final String zk = "localhost:2184";
         final Properties prop = new Properties();
         prop.setProperty("zookeeperServers", zk);
-        final ServiceConfiguration serviceConfig = ServiceConfigurationLoader.create(prop);
+        final ServiceConfiguration serviceConfig = PulsarConfigurationLoader.create(prop, ServiceConfiguration.class);
         assertNotNull(serviceConfig);
         assertEquals(serviceConfig.getZookeeperServers(), zk);
     }
 
     @Test
-    public void testServiceConfiguraitonComplete() throws Exception {
+    public void testPulsarConfiguraitonComplete() throws Exception {
         final String zk = "localhost:2184";
         final Properties prop = new Properties();
         prop.setProperty("zookeeperServers", zk);
-        final ServiceConfiguration serviceConfig = ServiceConfigurationLoader.create(prop);
-        assertEquals(serviceConfig.getZookeeperServers(), zk);
+        final ServiceConfiguration serviceConfig = PulsarConfigurationLoader.create(prop, ServiceConfiguration.class);
+        try {
+            isComplete(serviceConfig);
+            fail("it should fail as config is not complete");
+        } catch (IllegalArgumentException e) {
+            // Ok
+        }
     }
 
     @Test

pulsar-broker/src/main/java/com/yahoo/pulsar/PulsarBrokerStarter.java

@@ -15,8 +15,8 @@
  */
 package com.yahoo.pulsar;
 
-import static com.yahoo.pulsar.broker.ServiceConfigurationLoader.create;
-import static com.yahoo.pulsar.broker.ServiceConfigurationLoader.isComplete;
+import static com.yahoo.pulsar.common.configuration.PulsarConfigurationLoader.create;
+import static com.yahoo.pulsar.common.configuration.PulsarConfigurationLoader.isComplete;
 
 import java.io.FileInputStream;
 
@@ -33,7 +33,7 @@ public class PulsarBrokerStarter {
     private static ServiceConfiguration loadConfig(String configFile) throws Exception {
         SLF4JBridgeHandler.removeHandlersForRootLogger();
         SLF4JBridgeHandler.install();
-        ServiceConfiguration config = create((new FileInputStream(configFile)));
+        ServiceConfiguration config = create((new FileInputStream(configFile)), ServiceConfiguration.class);
         // it validates provided configuration is completed
         isComplete(config);
         return config;

pulsar-broker/src/main/java/com/yahoo/pulsar/PulsarStandaloneStarter.java

@@ -29,9 +29,9 @@
 import com.google.common.collect.Sets;
 import com.yahoo.pulsar.broker.PulsarService;
 import com.yahoo.pulsar.broker.ServiceConfiguration;
-import com.yahoo.pulsar.broker.ServiceConfigurationLoader;
 import com.yahoo.pulsar.client.admin.PulsarAdmin;
 import com.yahoo.pulsar.client.admin.PulsarAdminException;
+import com.yahoo.pulsar.common.configuration.PulsarConfigurationLoader;
 import com.yahoo.pulsar.common.policies.data.ClusterData;
 import com.yahoo.pulsar.common.policies.data.PropertyAdmin;
 import com.yahoo.pulsar.zookeeper.LocalBookkeeperEnsemble;
@@ -96,8 +96,8 @@ public PulsarStandaloneStarter(String[] args) throws Exception {
             return;
         }
 
-        this.config = ServiceConfigurationLoader.create((new FileInputStream(configFile)));
-        ServiceConfigurationLoader.isComplete(config);
+        this.config = PulsarConfigurationLoader.create((new FileInputStream(configFile)), ServiceConfiguration.class);
+        PulsarConfigurationLoader.isComplete(config);
         // Set ZK server's host to localhost
         config.setZookeeperServers("127.0.0.1:" + zkPort);
         config.setGlobalZookeeperServers("127.0.0.1:" + zkPort);

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/BacklogQuotaManagerTest.java

@@ -15,12 +15,10 @@
  */
 package com.yahoo.pulsar.broker.service;
 
-import static com.yahoo.pulsar.broker.ServiceConfigurationLoader.create;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 
 import java.net.URL;
-import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
@@ -77,7 +75,7 @@ void setup() throws Exception {
             bkEnsemble.start();
 
             // start pulsar service
-            config = create(new Properties(System.getProperties()));
+            config = new ServiceConfiguration();
             config.setZookeeperServers("127.0.0.1" + ":" + ZOOKEEPER_PORT);
             config.setWebServicePort(BROKER_WEBSERVICE_PORT);
             config.setClusterName("usc");

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/ReplicatorTestBase.java

@@ -15,13 +15,11 @@
  */
 package com.yahoo.pulsar.broker.service;
 
-import static com.yahoo.pulsar.broker.ServiceConfigurationLoader.create;
 import static org.testng.Assert.assertEquals;
 
 import java.net.URL;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -104,7 +102,7 @@ void setup() throws Exception {
         // NOTE: we have to instantiate a new copy of System.getProperties() to make sure pulsar1 and pulsar2 have
         // completely
         // independent config objects instead of referring to the same properties object
-        ServiceConfiguration config1 = create(new Properties(System.getProperties()));
+        ServiceConfiguration config1 = new ServiceConfiguration();
         config1.setClusterName("r1");
         config1.setWebServicePort(webServicePort1);
         config1.setZookeeperServers("127.0.0.1:" + zkPort1);
@@ -128,7 +126,7 @@ void setup() throws Exception {
         bkEnsemble2.start();
 
         int webServicePort2 = PortManager.nextFreePort();
-        config2 = create(new Properties(System.getProperties()));
+        config2 = new ServiceConfiguration();
         config2.setClusterName("r2");
         config2.setWebServicePort(webServicePort2);
         config2.setZookeeperServers("127.0.0.1:" + zkPort2);
@@ -152,7 +150,7 @@ void setup() throws Exception {
         bkEnsemble3.start();
 
         int webServicePort3 = PortManager.nextFreePort();
-        config3 = create(new Properties(System.getProperties()));
+        config3 = new ServiceConfiguration();
         config3.setClusterName("r3");
         config3.setWebServicePort(webServicePort3);
         config3.setZookeeperServers("127.0.0.1:" + zkPort3);

pulsar-broker/src/test/java/com/yahoo/pulsar/common/naming/ServiceConfigurationTest.java

@@ -31,7 +31,7 @@
 import org.testng.annotations.Test;
 
 import com.yahoo.pulsar.broker.ServiceConfiguration;
-import com.yahoo.pulsar.broker.ServiceConfigurationLoader;
+import com.yahoo.pulsar.common.configuration.PulsarConfigurationLoader;
 
 /**
  * 
@@ -51,7 +51,7 @@ public void testInit() throws Exception {
         final String zookeeperServer = "localhost:2184";
         final int brokerServicePort = 1000;
         InputStream newStream = updateProp(zookeeperServer, String.valueOf(brokerServicePort), "ns1,ns2");
-        final ServiceConfiguration config = ServiceConfigurationLoader.create(newStream);
+        final ServiceConfiguration config = PulsarConfigurationLoader.create(newStream, ServiceConfiguration.class);
         assertTrue(isNotBlank(config.getZookeeperServers()));
         assertTrue(config.getBrokerServicePort() == brokerServicePort);
         assertEquals(config.getBootstrapNamespaces().get(1), "ns2");
@@ -66,7 +66,7 @@ public void testInit() throws Exception {
     public void testInitFailure() throws Exception {
         final String zookeeperServer = "localhost:2184";
         InputStream newStream = updateProp(zookeeperServer, String.valueOf("invalid-string"), null);
-        ServiceConfigurationLoader.create(newStream);
+        PulsarConfigurationLoader.create(newStream, ServiceConfiguration.class);
     }
 
     private InputStream updateProp(String zookeeperServer, String brokerServicePort, String namespace)