diff --git hbase-hadoop-compat/pom.xml hbase-hadoop-compat/pom.xml index 05b06fc..796194b 100644 --- hbase-hadoop-compat/pom.xml +++ hbase-hadoop-compat/pom.xml @@ -58,6 +58,14 @@ + + com.google.inject + guice + + + com.google.inject.extensions + guice-assistedinject + diff --git hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/HadoopCompatPlugin.java hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/HadoopCompatPlugin.java new file mode 100644 index 0000000..9e60697 --- /dev/null +++ hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/HadoopCompatPlugin.java @@ -0,0 +1,35 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase; + +import com.google.inject.Module; + +import java.util.Collection; +import java.util.Map; + +/** + * + */ +public interface HadoopCompatPlugin { + + public String getName(); + + public Map getModules(); + +} diff --git hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/HadoopCompatPluginFactory.java hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/HadoopCompatPluginFactory.java new file mode 100644 index 0000000..14e83fa --- /dev/null +++ hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/HadoopCompatPluginFactory.java @@ -0,0 +1,49 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase; + +import com.google.inject.Module; + +import java.util.Collection; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.ServiceLoader; + +/** + * + */ +public class HadoopCompatPluginFactory { + + /** + * Get the singleton instance of ReplicationMetricsSource + * + * @return the singleton + */ + public static synchronized Map getModules() { + Map modules = new HashMap(); + for (HadoopCompatPlugin plugin : ServiceLoader.load(HadoopCompatPlugin.class)) { + modules.putAll(plugin.getModules()); + + } + return modules; + } + +} diff --git hbase-hadoop-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceFactoryTest.java hbase-hadoop-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceFactoryTest.java deleted file mode 100644 index d82e628..0000000 --- hbase-hadoop-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceFactoryTest.java +++ /dev/null @@ -1,34 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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 org.apache.hadoop.hbase.replication.regionserver.metrics; - -import org.junit.Test; - -/** - * Test for the ReplicationMetricsSourceFactory - */ -public class ReplicationMetricsSourceFactoryTest { - - @Test(expected=RuntimeException.class) - public void testGetInstanceNoHadoopCompat() throws Exception { - //This should throw an exception because there is no compat lib on the class path. - ReplicationMetricsSourceFactory.getInstance(); - - } -} diff --git hbase-hadoop-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceTest.java hbase-hadoop-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceTest.java new file mode 100644 index 0000000..2e769a4 --- /dev/null +++ hbase-hadoop-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceTest.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.replication.regionserver.metrics; + +import com.google.inject.Guice; +import org.apache.hadoop.hbase.HadoopCompatPluginFactory; +import org.junit.Test; + +/** + * Test for the ReplicationMetricsSource where there's no hadoop compat backing classes. + */ +public class ReplicationMetricsSourceTest { + + @Test(expected=RuntimeException.class) + public void testGetInstanceNoHadoopCompat() throws Exception { + //This should throw an exception because there is no compat lib on the class path. + Guice.createInjector(HadoopCompatPluginFactory.getModules().values()) + .getInstance(ReplicationMetricsSource.class); + } +} diff --git hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/HadoopOneCompatPlugin.java hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/HadoopOneCompatPlugin.java new file mode 100644 index 0000000..c6fc6e3 --- /dev/null +++ hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/HadoopOneCompatPlugin.java @@ -0,0 +1,47 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase; + +import com.google.inject.Module; +import org.apache.hadoop.hbase.guice.ReplicationMetricsModule; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * + */ +public class HadoopOneCompatPlugin implements HadoopCompatPlugin{ + + @Override + public String getName() { + return this.getClass().toString(); + } + + @Override + public Map getModules() { + Map modules = new HashMap(); + modules.put("ReplicationMetrics", new ReplicationMetricsModule()); + return modules; + + } +} diff --git hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/guice/ReplicationMetricsModule.java hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/guice/ReplicationMetricsModule.java new file mode 100644 index 0000000..61b4d7d --- /dev/null +++ hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/guice/ReplicationMetricsModule.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSource; +import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSourceImpl; + +import javax.inject.Singleton; + +/** + * + */ +public class ReplicationMetricsModule extends AbstractModule { + + @Override + protected void configure() { + bind(ReplicationMetricsSource.class).to(ReplicationMetricsSourceImpl.class).in(Singleton.class); + } +} diff --git hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.HadoopCompatPlugin hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.HadoopCompatPlugin new file mode 100644 index 0000000..7946bca --- /dev/null +++ hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.HadoopCompatPlugin @@ -0,0 +1 @@ +org.apache.hadoop.hbase.HadoopOneCompatPlugin \ No newline at end of file diff --git hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSource hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSource deleted file mode 100644 index bb64ad5..0000000 --- hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSource +++ /dev/null @@ -1 +0,0 @@ -org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSourceImpl \ No newline at end of file diff --git hbase-hadoop1-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceImplTest.java hbase-hadoop1-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceImplTest.java index 717aca5..b9d2ec3 100644 --- hbase-hadoop1-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceImplTest.java +++ hbase-hadoop1-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceImplTest.java @@ -18,18 +18,24 @@ package org.apache.hadoop.hbase.replication.regionserver.metrics; +import com.google.inject.Guice; +import org.apache.hadoop.hbase.HadoopCompatPluginFactory; import org.junit.Test; import static org.junit.Assert.assertTrue; -/** - * Test to make sure that ReplicationMetricsSourceImpl is hooked up to ServiceLoader - */ +/** Test for ReplicationMetricsSourceImpl */ public class ReplicationMetricsSourceImplTest { + /** + * Make sure that the guice is wired up. + * @throws Exception + */ @Test public void testGetInstance() throws Exception { - ReplicationMetricsSource rms = ReplicationMetricsSourceFactory.getInstance(); + ReplicationMetricsSource rms = + Guice.createInjector(HadoopCompatPluginFactory.getModules().values()) + .getInstance(ReplicationMetricsSource.class); assertTrue(rms instanceof ReplicationMetricsSourceImpl); } } diff --git hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/HadoopTwoCompatPlugin.java hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/HadoopTwoCompatPlugin.java new file mode 100644 index 0000000..60e10b6 --- /dev/null +++ hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/HadoopTwoCompatPlugin.java @@ -0,0 +1,47 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase; + +import com.google.inject.Module; +import org.apache.hadoop.hbase.guice.ReplicationMetricsModule; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * + */ +public class HadoopTwoCompatPlugin implements HadoopCompatPlugin{ + + @Override + public String getName() { + return this.getClass().toString(); + } + + @Override + public Map getModules() { + Map modules = new HashMap(); + modules.put("ReplicationMetrics", new ReplicationMetricsModule()); + return modules; + + } +} diff --git hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/guice/ReplicationMetricsModule.java hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/guice/ReplicationMetricsModule.java new file mode 100644 index 0000000..61b4d7d --- /dev/null +++ hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/guice/ReplicationMetricsModule.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSource; +import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSourceImpl; + +import javax.inject.Singleton; + +/** + * + */ +public class ReplicationMetricsModule extends AbstractModule { + + @Override + protected void configure() { + bind(ReplicationMetricsSource.class).to(ReplicationMetricsSourceImpl.class).in(Singleton.class); + } +} diff --git hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.HadoopCompatPlugin hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.HadoopCompatPlugin new file mode 100644 index 0000000..490cabf --- /dev/null +++ hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.HadoopCompatPlugin @@ -0,0 +1 @@ +org.apache.hadoop.hbase.HadoopTwoCompatPlugin \ No newline at end of file diff --git hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSource hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSource deleted file mode 100644 index bb64ad5..0000000 --- hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSource +++ /dev/null @@ -1 +0,0 @@ -org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSourceImpl \ No newline at end of file diff --git hbase-hadoop2-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceImplTest.java hbase-hadoop2-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceImplTest.java index af31c6e..6640307 100644 --- hbase-hadoop2-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceImplTest.java +++ hbase-hadoop2-compat/src/test/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationMetricsSourceImplTest.java @@ -18,6 +18,9 @@ package org.apache.hadoop.hbase.replication.regionserver.metrics; +import com.google.inject.Guice; +import org.apache.hadoop.hbase.HadoopCompatPlugin; +import org.apache.hadoop.hbase.HadoopCompatPluginFactory; import org.junit.Test; import static org.junit.Assert.assertTrue; @@ -25,9 +28,17 @@ import static org.junit.Assert.assertTrue; /** Test for ReplicationMetricsSourceImpl */ public class ReplicationMetricsSourceImplTest { + /** + * Make sure that the guice is wired up. + * @throws Exception + */ @Test public void testGetInstance() throws Exception { - ReplicationMetricsSource rms = ReplicationMetricsSourceFactory.getInstance(); + + + ReplicationMetricsSource rms = + Guice.createInjector(HadoopCompatPluginFactory.getModules().values()) + .getInstance(ReplicationMetricsSource.class); assertTrue(rms instanceof ReplicationMetricsSourceImpl); } } diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/LocalHBaseCluster.java hbase-server/src/main/java/org/apache/hadoop/hbase/LocalHBaseCluster.java index f550c7a..b3dc64a 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/LocalHBaseCluster.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/LocalHBaseCluster.java @@ -25,15 +25,14 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.client.HBaseAdmin; import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.security.User; -import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread; import org.apache.hadoop.hbase.util.Threads; @@ -41,6 +40,9 @@ import java.util.concurrent.CopyOnWriteArrayList; import org.apache.hadoop.hbase.master.HMaster; import org.apache.hadoop.hbase.util.JVMClusterUtil; +import javax.inject.Inject; +import javax.inject.Provider; + /** * This class creates a single process HBase cluster. One thread is created for * a master and one per region server. @@ -71,58 +73,8 @@ public class LocalHBaseCluster { /** 'local:' */ public static final String LOCAL_COLON = LOCAL + ":"; private final Configuration conf; - private final Class masterClass; - private final Class regionServerClass; - - /** - * Constructor. - * @param conf - * @throws IOException - */ - public LocalHBaseCluster(final Configuration conf) - throws IOException { - this(conf, DEFAULT_NO); - } - - /** - * Constructor. - * @param conf Configuration to use. Post construction has the master's - * address. - * @param noRegionServers Count of regionservers to start. - * @throws IOException - */ - public LocalHBaseCluster(final Configuration conf, final int noRegionServers) - throws IOException { - this(conf, 1, noRegionServers, getMasterImplementation(conf), - getRegionServerImplementation(conf)); - } - - /** - * Constructor. - * @param conf Configuration to use. Post construction has the active master - * address. - * @param noMasters Count of masters to start. - * @param noRegionServers Count of regionservers to start. - * @throws IOException - */ - public LocalHBaseCluster(final Configuration conf, final int noMasters, - final int noRegionServers) - throws IOException { - this(conf, noMasters, noRegionServers, getMasterImplementation(conf), - getRegionServerImplementation(conf)); - } - - @SuppressWarnings("unchecked") - private static Class getRegionServerImplementation(final Configuration conf) { - return (Class)conf.getClass(HConstants.REGION_SERVER_IMPL, - HRegionServer.class); - } - - @SuppressWarnings("unchecked") - private static Class getMasterImplementation(final Configuration conf) { - return (Class)conf.getClass(HConstants.MASTER_IMPL, - HMaster.class); - } + private final Provider masterFactory; + private final Provider regionServerFactory; /** * Constructor. @@ -130,87 +82,84 @@ public class LocalHBaseCluster { * address. * @param noMasters Count of masters to start. * @param noRegionServers Count of regionservers to start. - * @param masterClass - * @param regionServerClass * @throws IOException */ @SuppressWarnings("unchecked") - public LocalHBaseCluster(final Configuration conf, final int noMasters, - final int noRegionServers, final Class masterClass, - final Class regionServerClass) + @Inject + public LocalHBaseCluster(final Configuration conf, + Provider masterFactory, + Provider regionServerFactory, + @Assisted("noMasters") final int noMasters, + @Assisted("noRegionServers") final int noRegionServers) throws IOException { this.conf = conf; + this.masterFactory = masterFactory; + this.regionServerFactory = regionServerFactory; + // Always have masters and regionservers come up on port '0' so we don't // clash over default ports. conf.set(HConstants.MASTER_PORT, "0"); conf.set(HConstants.REGIONSERVER_PORT, "0"); - this.masterClass = (Class) - conf.getClass(HConstants.MASTER_IMPL, masterClass); + // Start the HMasters. for (int i = 0; i < noMasters; i++) { - addMaster(new Configuration(conf), i); + addMaster(masterFactory, i); } // Start the HRegionServers. - this.regionServerClass = - (Class)conf.getClass(HConstants.REGION_SERVER_IMPL, - regionServerClass); for (int i = 0; i < noRegionServers; i++) { - addRegionServer(new Configuration(conf), i); + addRegionServer(regionServerFactory, i); } } public JVMClusterUtil.RegionServerThread addRegionServer() throws IOException { - return addRegionServer(new Configuration(conf), this.regionThreads.size()); + return addRegionServer(this.regionServerFactory, this.regionThreads.size()); } public JVMClusterUtil.RegionServerThread addRegionServer( - Configuration config, final int index) + Provider hrsf, final int index) throws IOException { // Create each regionserver with its own Configuration instance so each has // its HConnection instance rather than share (see HBASE_INSTANCES down in // the guts of HConnectionManager. JVMClusterUtil.RegionServerThread rst = - JVMClusterUtil.createRegionServerThread(config, - this.regionServerClass, index); + JVMClusterUtil.createRegionServerThread(hrsf, new Configuration(conf), index); this.regionThreads.add(rst); return rst; } - public JVMClusterUtil.RegionServerThread addRegionServer( - final Configuration config, final int index, User user) + public JVMClusterUtil.RegionServerThread addRegionServer(final int index, User user) throws IOException, InterruptedException { return user.runAs( new PrivilegedExceptionAction() { public JVMClusterUtil.RegionServerThread run() throws Exception { - return addRegionServer(config, index); + return addRegionServer(regionServerFactory, index); } }); } public JVMClusterUtil.MasterThread addMaster() throws IOException { - return addMaster(new Configuration(conf), this.masterThreads.size()); + return addMaster(this.masterFactory, this.masterThreads.size()); } - public JVMClusterUtil.MasterThread addMaster(Configuration c, final int index) + public JVMClusterUtil.MasterThread addMaster(Provider mf, final int index) throws IOException { // Create each master with its own Configuration instance so each has // its HConnection instance rather than share (see HBASE_INSTANCES down in // the guts of HConnectionManager. - JVMClusterUtil.MasterThread mt = JVMClusterUtil.createMasterThread(c, - (Class) conf.getClass(HConstants.MASTER_IMPL, this.masterClass), index); + JVMClusterUtil.MasterThread mt = JVMClusterUtil.createMasterThread(mf, new Configuration(conf), index); this.masterThreads.add(mt); return mt; } public JVMClusterUtil.MasterThread addMaster( - final Configuration c, final int index, User user) + final int index, User user) throws IOException, InterruptedException { return user.runAs( new PrivilegedExceptionAction() { public JVMClusterUtil.MasterThread run() throws Exception { - return addMaster(c, index); + return addMaster(masterFactory, index); } }); } @@ -439,20 +388,4 @@ public class LocalHBaseCluster { boolean mode = c.getBoolean(HConstants.CLUSTER_DISTRIBUTED, HConstants.DEFAULT_CLUSTER_DISTRIBUTED); return(mode == HConstants.CLUSTER_IS_LOCAL); } - - /** - * Test things basically work. - * @param args - * @throws IOException - */ - public static void main(String[] args) throws IOException { - Configuration conf = HBaseConfiguration.create(); - LocalHBaseCluster cluster = new LocalHBaseCluster(conf); - cluster.startup(); - HBaseAdmin admin = new HBaseAdmin(conf); - HTableDescriptor htd = - new HTableDescriptor(Bytes.toBytes(cluster.getClass().getName())); - admin.createTable(htd); - cluster.shutdown(); - } } diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/CopyingConfModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/CopyingConfModule.java new file mode 100644 index 0000000..9dd44dd --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/CopyingConfModule.java @@ -0,0 +1,44 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.Provides; +import org.apache.hadoop.conf.Configuration; + +/** + * + */ +public class CopyingConfModule extends AbstractModule { + + private Configuration conf; + + public CopyingConfModule(Configuration conf) { + this.conf = conf; + } + + @Override + protected void configure() { + } + + @Provides + public Configuration createConf() { + return new Configuration(conf); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ExplicitConfModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ExplicitConfModule.java new file mode 100644 index 0000000..555b846 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ExplicitConfModule.java @@ -0,0 +1,40 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import org.apache.hadoop.conf.Configuration; + +/** + * + */ +public class ExplicitConfModule extends AbstractModule { + + private Configuration configuration; + + public ExplicitConfModule(Configuration configuration) { + this.configuration = configuration; + } + + @Override + protected void configure() { + bind(Configuration.class).toInstance(this.configuration); + + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/HBaseGuice.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/HBaseGuice.java new file mode 100644 index 0000000..219b933 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/HBaseGuice.java @@ -0,0 +1,69 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.Guice; +import com.google.inject.Injector; +import com.google.inject.Module; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HadoopCompatPluginFactory; + +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; + +/** + * + */ +public class HBaseGuice { + + public static Injector createLocalHBaseInjector(Configuration conf) { + Collection modules = makeLocalHBaseModules(conf).values(); + return Guice.createInjector(modules); + } + + public static Injector createInjector(Configuration conf) { + Collection modules = makeDefaultModules(conf).values(); + return Guice.createInjector(modules); + } + + public static Map makeDefaultModules(Configuration conf) { + Map modules = new HashMap(); + modules.put("Conf", new ExplicitConfModule(conf)); + modules.put("Sleeper", new SleeperModule()); + modules.put("Replication", new ReplicationModule()); + modules.put("ReplicationZookeeper", new ReplicationZookeeperModule()); + modules.put("ReplicationSourceManager", new ReplicationSourceManagerModule()); + modules.put("ReplicationSource", new ReplicationSourceModule()); + modules.put("ReplicationSink", new ReplicationSinkModule()); + modules.put("RpcServer", new RpcServerModule()); + modules.put("RegionServerAccounting", new RegionServerAccountingModule()); + modules.put("ZooKeeperWatcher", new ZooKeeperWatcherModule()); + modules.putAll(HadoopCompatPluginFactory.getModules()); + return modules; + } + + public static Map makeLocalHBaseModules(Configuration conf) { + Map modules = makeDefaultModules(conf); + modules.put("Conf", new CopyingConfModule(conf)); + modules.put("LocalHBaseCluster", new LocalHBaseClusterModule()); + return modules; + } + +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/LocalHBaseClusterFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/LocalHBaseClusterFactory.java new file mode 100644 index 0000000..a788cb8 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/LocalHBaseClusterFactory.java @@ -0,0 +1,32 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.assistedinject.Assisted; +import org.apache.hadoop.hbase.LocalHBaseCluster; + +/** + * + */ +public interface LocalHBaseClusterFactory { + + public LocalHBaseCluster create(@Assisted("noMasters") int noMasters, + @Assisted("noRegionServers") int noRegionservers); + +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/LocalHBaseClusterModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/LocalHBaseClusterModule.java new file mode 100644 index 0000000..079024f --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/LocalHBaseClusterModule.java @@ -0,0 +1,41 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.assistedinject.FactoryModuleBuilder; +import org.apache.hadoop.hbase.LocalHBaseCluster; +import org.apache.hadoop.hbase.master.HMaster; +import org.apache.hadoop.hbase.master.LocalHMaster; +import org.apache.hadoop.hbase.regionserver.HRegionServer; + +/** + * + */ +public class LocalHBaseClusterModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(LocalHBaseCluster.class, LocalHBaseCluster.class) + .build(LocalHBaseClusterFactory.class)); + + bind(HMaster.class).to(LocalHMaster.class); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RegionServerAccountingModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RegionServerAccountingModule.java new file mode 100644 index 0000000..9ff3bf9 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RegionServerAccountingModule.java @@ -0,0 +1,33 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import org.apache.hadoop.hbase.regionserver.RegionServerAccounting; + +/** + * + */ +public class RegionServerAccountingModule extends AbstractModule { + + @Override + protected void configure() { +// bind(RegionServerAccounting.class).to(RegionServerAccounting.class); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationFactory.java new file mode 100644 index 0000000..00d006d --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationFactory.java @@ -0,0 +1,39 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.assistedinject.Assisted; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.Server; +import org.apache.hadoop.hbase.replication.regionserver.Replication; + +import java.io.IOException; + +/** + * + */ +public interface ReplicationFactory { + + public Replication create(@Assisted final Server server, + @Assisted final FileSystem fs, + @Assisted("logDir") final Path logDir, + @Assisted("oldLogDir") final Path oldLogDir) throws IOException; + +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationModule.java new file mode 100644 index 0000000..c2d6d25 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationModule.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.assistedinject.FactoryModuleBuilder; +import org.apache.hadoop.hbase.replication.regionserver.Replication; + +/** + * + */ +public class ReplicationModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(Replication.class, Replication.class) + .build(ReplicationFactory.class)); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSinkFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSinkFactory.java new file mode 100644 index 0000000..b38edc5 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSinkFactory.java @@ -0,0 +1,29 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import org.apache.hadoop.hbase.replication.regionserver.ReplicationSink; + +/** + * + */ +public interface ReplicationSinkFactory { + + public ReplicationSink create(); +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSinkModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSinkModule.java new file mode 100644 index 0000000..b5d4afc --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSinkModule.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.assistedinject.FactoryModuleBuilder; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationSink; + +/** + * + */ +public class ReplicationSinkModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(ReplicationSink.class, ReplicationSink.class) + .build(ReplicationSinkFactory.class)); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceFactory.java new file mode 100644 index 0000000..d7177c1 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceFactory.java @@ -0,0 +1,40 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.hbase.Stoppable; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceInterface; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * + */ +public interface ReplicationSourceFactory { + + ReplicationSourceInterface create( + final FileSystem fs, + final ReplicationSourceManager manager, + final Stoppable stopper, + final AtomicBoolean replicating, + final String peerClusterZnode); + +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceManagerFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceManagerFactory.java new file mode 100644 index 0000000..f7eb061 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceManagerFactory.java @@ -0,0 +1,42 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.assistedinject.Assisted; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.Stoppable; +import org.apache.hadoop.hbase.replication.ReplicationZookeeper; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * + */ +public interface ReplicationSourceManagerFactory { + + public ReplicationSourceManager create(final ReplicationZookeeper zkHelper, + final Stoppable stopper, + final FileSystem fs, + final AtomicBoolean replicating, + @Assisted("logDir") final Path logDir, + @Assisted("oldLogDir") final Path oldLogDir); + +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceManagerModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceManagerModule.java new file mode 100644 index 0000000..fa913ac --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceManagerModule.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.assistedinject.FactoryModuleBuilder; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; + +/** + * + */ +public class ReplicationSourceManagerModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(ReplicationSourceManager.class, ReplicationSourceManager.class) + .build(ReplicationSourceManagerFactory.class)); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceMetricsFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceMetricsFactory.java new file mode 100644 index 0000000..3b976e3 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceMetricsFactory.java @@ -0,0 +1,30 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationSourceMetrics; + +/** + * + */ +public interface ReplicationSourceMetricsFactory { + + public ReplicationSourceMetrics create(String peerClusterZnode); + +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceModule.java new file mode 100644 index 0000000..756d1ea --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationSourceModule.java @@ -0,0 +1,37 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.assistedinject.FactoryModuleBuilder; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationSource; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceInterface; + +/** + * + */ +public class ReplicationSourceModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(ReplicationSourceInterface.class, ReplicationSource.class) + .build(ReplicationSourceFactory.class)); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationZookeeperFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationZookeeperFactory.java new file mode 100644 index 0000000..5933de5 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationZookeeperFactory.java @@ -0,0 +1,35 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import org.apache.hadoop.hbase.Server; +import org.apache.hadoop.hbase.replication.ReplicationZookeeper; +import org.apache.zookeeper.KeeperException; + +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * + */ +public interface ReplicationZookeeperFactory { + + public ReplicationZookeeper create(final Server server, final AtomicBoolean replicating) throws + KeeperException; + +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationZookeeperModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationZookeeperModule.java new file mode 100644 index 0000000..76d8c45 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ReplicationZookeeperModule.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.assistedinject.FactoryModuleBuilder; +import org.apache.hadoop.hbase.replication.ReplicationZookeeper; + +/** + * + */ +public class ReplicationZookeeperModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(ReplicationZookeeper.class, ReplicationZookeeper.class) + .build(ReplicationZookeeperFactory.class)); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RpcServerDefaultFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RpcServerDefaultFactory.java new file mode 100644 index 0000000..71ee678 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RpcServerDefaultFactory.java @@ -0,0 +1,52 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.ipc.HBaseRPC; +import org.apache.hadoop.hbase.ipc.RpcServer; + +import java.io.IOException; + +/** + * + */ +public class RpcServerDefaultFactory implements RpcServerFactory { + + @Override + public RpcServer create(Object instance, + Class[] ifaces, + String bindAddress, + int port, + int numHandlers, + int metaHandlerCount, + boolean verbose, + Configuration conf, + int highPriorityLevel) throws IOException { + return HBaseRPC.getServer(instance, + ifaces, + bindAddress, + port, + numHandlers, + metaHandlerCount, + verbose, + conf, + highPriorityLevel); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RpcServerFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RpcServerFactory.java new file mode 100644 index 0000000..b026ac1 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RpcServerFactory.java @@ -0,0 +1,41 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.ipc.RpcServer; + +import java.io.IOException; + +/** + * + */ +public interface RpcServerFactory { + + public RpcServer create(final Object instance, + final Class[] ifaces, + final String bindAddress, + final int port, + final int numHandlers, + int metaHandlerCount, + final boolean verbose, + Configuration conf, + int highPriorityLevel) throws IOException; + +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RpcServerModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RpcServerModule.java new file mode 100644 index 0000000..b43a1d7 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/RpcServerModule.java @@ -0,0 +1,32 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; + +/** + * + */ +public class RpcServerModule extends AbstractModule { + + @Override + protected void configure() { + bind(RpcServerFactory.class).to(RpcServerDefaultFactory.class); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/SleeperFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/SleeperFactory.java new file mode 100644 index 0000000..af88ee8 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/SleeperFactory.java @@ -0,0 +1,30 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import org.apache.hadoop.hbase.Stoppable; +import org.apache.hadoop.hbase.util.Sleeper; + +/** + * + */ +public interface SleeperFactory { + + public Sleeper create(final int msgInterval, final Stoppable stoppable); +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/SleeperModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/SleeperModule.java new file mode 100644 index 0000000..b42b3f7 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/SleeperModule.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.assistedinject.FactoryModuleBuilder; +import org.apache.hadoop.hbase.util.Sleeper; + +/** + * + */ +public class SleeperModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(Sleeper.class, Sleeper.class) + .build(SleeperFactory.class)); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ZooKeeperWatcherFactory.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ZooKeeperWatcherFactory.java new file mode 100644 index 0000000..6999b5c --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ZooKeeperWatcherFactory.java @@ -0,0 +1,33 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import org.apache.hadoop.hbase.Abortable; +import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher; + +/** + * + */ +public interface ZooKeeperWatcherFactory { + + public ZooKeeperWatcher create(String descriptor, + Abortable abortable, + boolean canCreateBaseZNode); + +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ZooKeeperWatcherModule.java hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ZooKeeperWatcherModule.java new file mode 100644 index 0000000..74479e0 --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/guice/ZooKeeperWatcherModule.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.assistedinject.FactoryModuleBuilder; +import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher; + +/** + * + */ +public class ZooKeeperWatcherModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(ZooKeeperWatcher.class, ZooKeeperWatcher.class) + .build(ZooKeeperWatcherFactory.class)); + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java index dc38322..dbbb6f6 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java @@ -40,6 +40,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import javax.inject.Inject; import javax.management.ObjectName; import org.apache.commons.logging.Log; @@ -314,7 +315,8 @@ Server { * run in their own thread rather than within the context of the constructor. * @throws InterruptedException */ - public HMaster(final Configuration conf) + @Inject + public HMaster(final Configuration conf ) throws IOException, KeeperException, InterruptedException { this.conf = new Configuration(conf); // Disable the block cache on the master @@ -1126,7 +1128,7 @@ Server { */ protected RegionServerStartupResponse.Builder createConfigurationSubset() { RegionServerStartupResponse.Builder resp = addConfig( - RegionServerStartupResponse.newBuilder(), HConstants.HBASE_DIR); + RegionServerStartupResponse.newBuilder(), HConstants.HBASE_DIR); return addConfig(resp, "fs.default.name"); } @@ -2220,31 +2222,6 @@ Server { } /** - * Utility for constructing an instance of the passed HMaster class. - * @param masterClass - * @param conf - * @return HMaster instance. - */ - public static HMaster constructMaster(Class masterClass, - final Configuration conf) { - try { - Constructor c = - masterClass.getConstructor(Configuration.class); - return c.newInstance(conf); - } catch (InvocationTargetException ite) { - Throwable target = ite.getTargetException() != null? - ite.getTargetException(): ite; - if (target.getCause() != null) target = target.getCause(); - throw new RuntimeException("Failed construction of Master: " + - masterClass.toString(), target); - } catch (Exception e) { - throw new RuntimeException("Failed construction of Master: " + - masterClass.toString() + ((e.getCause() != null)? - e.getCause().getMessage(): ""), e); - } - } - - /** * @see org.apache.hadoop.hbase.master.HMasterCommandLine */ public static void main(String [] args) throws Exception { diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMasterCommandLine.java hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMasterCommandLine.java index 16a3cd8..3c0d85f 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMasterCommandLine.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMasterCommandLine.java @@ -23,6 +23,8 @@ import java.io.File; import java.io.IOException; import java.util.List; +import com.google.inject.Guice; +import com.google.inject.Injector; import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.GnuParser; import org.apache.commons.cli.Options; @@ -37,6 +39,10 @@ import org.apache.hadoop.hbase.LocalHBaseCluster; import org.apache.hadoop.hbase.MasterNotRunningException; import org.apache.hadoop.hbase.ZooKeeperConnectionException; import org.apache.hadoop.hbase.client.HBaseAdmin; +import org.apache.hadoop.hbase.guice.ExplicitConfModule; +import org.apache.hadoop.hbase.guice.HBaseGuice; +import org.apache.hadoop.hbase.guice.LocalHBaseClusterFactory; +import org.apache.hadoop.hbase.guice.LocalHBaseClusterModule; import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.util.JVMClusterUtil; import org.apache.hadoop.hbase.util.ServerCommandLine; @@ -143,13 +149,15 @@ public class HMasterCommandLine extends ServerCommandLine { Integer.toString(clientPort)); // Need to have the zk cluster shutdown when master is shutdown. // Run a subclass that does the zk cluster shutdown on its way out. - LocalHBaseCluster cluster = new LocalHBaseCluster(conf, 1, 1, - LocalHMaster.class, HRegionServer.class); + Injector injector = HBaseGuice.createLocalHBaseInjector(conf); + LocalHBaseCluster cluster = injector.getInstance(LocalHBaseClusterFactory.class).create(1,1); + ((LocalHMaster)cluster.getMaster(0)).setZKCluster(zooKeeperCluster); cluster.startup(); waitOnMasterThreads(cluster); } else { - HMaster master = HMaster.constructMaster(masterClass, conf); + Injector injector = HBaseGuice.createInjector(conf); + HMaster master = injector.getInstance(masterClass); if (master.isStopped()) { LOG.info("Won't bring the Master up as a shutdown is requested"); return -1; @@ -209,32 +217,4 @@ public class HMasterCommandLine extends ServerCommandLine { t.getRegionServer().stop("HMaster Aborted; Bringing down regions servers"); } } - - /* - * Version of master that will shutdown the passed zk cluster on its way out. - */ - public static class LocalHMaster extends HMaster { - private MiniZooKeeperCluster zkcluster = null; - - public LocalHMaster(Configuration conf) - throws IOException, KeeperException, InterruptedException { - super(conf); - } - - @Override - public void run() { - super.run(); - if (this.zkcluster != null) { - try { - this.zkcluster.shutdown(); - } catch (IOException e) { - e.printStackTrace(); - } - } - } - - void setZKCluster(final MiniZooKeeperCluster zkcluster) { - this.zkcluster = zkcluster; - } - } } diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/master/LocalHMaster.java hbase-server/src/main/java/org/apache/hadoop/hbase/master/LocalHMaster.java new file mode 100644 index 0000000..decde6e --- /dev/null +++ hbase-server/src/main/java/org/apache/hadoop/hbase/master/LocalHMaster.java @@ -0,0 +1,63 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.master; + +import com.google.inject.assistedinject.Assisted; +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster; +import org.apache.zookeeper.KeeperException; + +import javax.inject.Inject; +import java.io.IOException; + +/** + * + */ +/* +* Version of master that will shutdown the passed zk cluster on its way out. +*/ +@InterfaceAudience.Private +public class LocalHMaster extends HMaster { + + private MiniZooKeeperCluster zkcluster = null; + + + @Inject + LocalHMaster(Configuration conf) + throws IOException, KeeperException, InterruptedException { + super(conf); + } + + @Override + public void run() { + super.run(); + if (this.zkcluster != null) { + try { + this.zkcluster.shutdown(); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + + void setZKCluster(final MiniZooKeeperCluster zkcluster) { + this.zkcluster = zkcluster; + } +} diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java index e2bbb48..8a890fe 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java @@ -51,8 +51,10 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.ReentrantReadWriteLock; +import javax.inject.Inject; import javax.management.ObjectName; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.lang.mutable.MutableDouble; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -104,6 +106,10 @@ import org.apache.hadoop.hbase.executor.ExecutorService.ExecutorType; import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp; import org.apache.hadoop.hbase.filter.WritableByteArrayComparable; import org.apache.hadoop.hbase.fs.HFileSystem; +import org.apache.hadoop.hbase.guice.ReplicationFactory; +import org.apache.hadoop.hbase.guice.RpcServerFactory; +import org.apache.hadoop.hbase.guice.SleeperFactory; +import org.apache.hadoop.hbase.guice.ZooKeeperWatcherFactory; import org.apache.hadoop.hbase.io.hfile.BlockCache; import org.apache.hadoop.hbase.io.hfile.CacheConfig; import org.apache.hadoop.hbase.io.hfile.CacheStats; @@ -435,7 +441,8 @@ public class HRegionServer implements ClientProtocol, * The lease timeout period for client scanners (milliseconds). */ private final int scannerLeaseTimeoutPeriod; - + private ReplicationFactory replicationFactory; + private ZooKeeperWatcherFactory zooKeeperWatcherFactory; /** * Starts a HRegionServer at the default location @@ -444,8 +451,16 @@ public class HRegionServer implements ClientProtocol, * @throws IOException * @throws InterruptedException */ - public HRegionServer(Configuration conf) - throws IOException, InterruptedException { + @Inject + public HRegionServer(Configuration conf, + SleeperFactory sleeperFactory, + RpcServerFactory rpcServerFactory, + ReplicationFactory replicationFactory, + RegionServerAccounting rsAccounting, + ZooKeeperWatcherFactory zooKeeperWatcherFactory) + throws IOException, InterruptedException { + this.replicationFactory = replicationFactory; + this.zooKeeperWatcherFactory = zooKeeperWatcherFactory; this.fsOk = true; this.conf = conf; // Set how many times to retry talking to another server over HConnection. @@ -464,7 +479,7 @@ public class HRegionServer implements ClientProtocol, 10 * 1000); this.msgInterval = conf.getInt("hbase.regionserver.msginterval", 3 * 1000); - this.sleeper = new Sleeper(this.msgInterval, this); + this.sleeper = sleeperFactory.create(this.msgInterval, this); this.maxScannerResultSize = conf.getLong( HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY, @@ -499,10 +514,11 @@ public class HRegionServer implements ClientProtocol, throw new IllegalArgumentException("Failed resolve of " + initialIsa); } - this.rpcServer = HBaseRPC.getServer(this, - new Class[]{ClientProtocol.class, - AdminProtocol.class, HBaseRPCErrorHandler.class, - OnlineRegions.class}, + this.rpcServer = rpcServerFactory.create(this, + new Class[]{ ClientProtocol.class, + AdminProtocol.class, + HBaseRPCErrorHandler.class, + OnlineRegions.class}, initialIsa.getHostName(), // BindAddress is IP we got for this server. initialIsa.getPort(), conf.getInt("hbase.regionserver.handler.count", 10), @@ -519,8 +535,8 @@ public class HRegionServer implements ClientProtocol, // login the server principal (if using secure Hadoop) User.login(this.conf, "hbase.regionserver.keytab.file", "hbase.regionserver.kerberos.principal", this.isa.getHostName()); - regionServerAccounting = new RegionServerAccounting(); - cacheConfig = new CacheConfig(conf); + this.regionServerAccounting = rsAccounting; + this.cacheConfig = new CacheConfig(conf); } /** @@ -667,8 +683,8 @@ public class HRegionServer implements ClientProtocol, */ private void initializeZooKeeper() throws IOException, InterruptedException { // Open connection to zookeeper and set primary watcher - this.zooKeeper = new ZooKeeperWatcher(conf, REGIONSERVER + ":" + - this.isa.getPort(), this); + this.zooKeeper = zooKeeperWatcherFactory.create(REGIONSERVER + ":" + + this.isa.getPort(), this, false); // Create the master address manager, register with zk, and start it. Then // block until a master is available. No point in starting up if no master @@ -1282,7 +1298,7 @@ public class HRegionServer implements ClientProtocol, // Instantiate replication manager if replication enabled. Pass it the // log directories. - createNewReplicationInstance(conf, this, this.fs, logdir, oldLogDir); + createNewReplicationInstance(logdir, oldLogDir); return instantiateHLog(logdir, oldLogDir); } @@ -2197,39 +2213,16 @@ public class HRegionServer implements ClientProtocol, /** * Load the replication service objects, if any */ - static private void createNewReplicationInstance(Configuration conf, - HRegionServer server, FileSystem fs, Path logDir, Path oldLogDir) throws IOException{ + private void createNewReplicationInstance( Path logDir, Path oldLogDir) throws IOException{ // If replication is not enabled, then return immediately. if (!conf.getBoolean(HConstants.REPLICATION_ENABLE_KEY, false)) { return; } - // read in the name of the source replication class from the config file. - String sourceClassname = conf.get(HConstants.REPLICATION_SOURCE_SERVICE_CLASSNAME, - HConstants.REPLICATION_SERVICE_CLASSNAME_DEFAULT); - - // read in the name of the sink replication class from the config file. - String sinkClassname = conf.get(HConstants.REPLICATION_SINK_SERVICE_CLASSNAME, - HConstants.REPLICATION_SERVICE_CLASSNAME_DEFAULT); + this.replicationSinkHandler = this.replicationFactory.create(this, fs, logDir, oldLogDir); + this.replicationSourceHandler = (ReplicationSourceService) this.replicationSinkHandler; - // If both the sink and the source class names are the same, then instantiate - // only one object. - if (sourceClassname.equals(sinkClassname)) { - server.replicationSourceHandler = (ReplicationSourceService) - newReplicationInstance(sourceClassname, - conf, server, fs, logDir, oldLogDir); - server.replicationSinkHandler = (ReplicationSinkService) - server.replicationSourceHandler; - } - else { - server.replicationSourceHandler = (ReplicationSourceService) - newReplicationInstance(sourceClassname, - conf, server, fs, logDir, oldLogDir); - server.replicationSinkHandler = (ReplicationSinkService) - newReplicationInstance(sinkClassname, - conf, server, fs, logDir, oldLogDir); - } } static private ReplicationService newReplicationInstance(String classname, diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServerCommandLine.java hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServerCommandLine.java index c0cdde5..d55c2ef 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServerCommandLine.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServerCommandLine.java @@ -19,6 +19,8 @@ */ package org.apache.hadoop.hbase.regionserver; +import com.google.inject.Guice; +import com.google.inject.Injector; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -26,6 +28,8 @@ import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.LocalHBaseCluster; +import org.apache.hadoop.hbase.guice.ExplicitConfModule; +import org.apache.hadoop.hbase.guice.HBaseGuice; import org.apache.hadoop.hbase.util.ServerCommandLine; /** @@ -59,7 +63,8 @@ public class HRegionServerCommandLine extends ServerCommandLine { + HConstants.CLUSTER_DISTRIBUTED + " is false"); } else { logJVMInfo(); - HRegionServer hrs = HRegionServer.constructRegionServer(regionServerClass, conf); + Injector injector = HBaseGuice.createInjector(conf); + HRegionServer hrs = injector.getInstance(regionServerClass); HRegionServer.startRegionServer(hrs); } return 0; diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java index 6eaa51f..555cebb 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java @@ -32,6 +32,7 @@ import java.util.TreeMap; import java.util.TreeSet; import java.util.concurrent.atomic.AtomicBoolean; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience; @@ -49,6 +50,8 @@ import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.ConnectionLossException; import org.apache.zookeeper.KeeperException.SessionExpiredException; +import javax.inject.Inject; + /** * This class serves as a helper for all things related to zookeeper in * replication. @@ -139,7 +142,9 @@ public class ReplicationZookeeper implements Closeable{ * @throws IOException * @throws KeeperException */ - public ReplicationZookeeper(final Server server, final AtomicBoolean replicating) + @Inject + public ReplicationZookeeper(@Assisted final Server server, + @Assisted final AtomicBoolean replicating) throws IOException, KeeperException { this.abortable = server; this.zookeeper = server.getZooKeeper(); diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java index 250ea86..d693a57 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java @@ -24,6 +24,7 @@ import java.util.NavigableMap; import java.util.TreeMap; import java.util.concurrent.atomic.AtomicBoolean; +import com.google.inject.assistedinject.Assisted; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; @@ -32,6 +33,9 @@ import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.Server; +import org.apache.hadoop.hbase.guice.ReplicationSinkFactory; +import org.apache.hadoop.hbase.guice.ReplicationSourceManagerFactory; +import org.apache.hadoop.hbase.guice.ReplicationZookeeperFactory; import org.apache.hadoop.hbase.regionserver.ReplicationSourceService; import org.apache.hadoop.hbase.regionserver.ReplicationSinkService; import org.apache.hadoop.hbase.regionserver.wal.HLog; @@ -43,6 +47,8 @@ import org.apache.hadoop.hbase.replication.master.ReplicationLogCleaner; import org.apache.hadoop.hbase.util.Bytes; import org.apache.zookeeper.KeeperException; +import javax.inject.Inject; + import static org.apache.hadoop.hbase.HConstants.HBASE_MASTER_LOGCLEANER_PLUGINS; import static org.apache.hadoop.hbase.HConstants.REPLICATION_ENABLE_KEY; import static org.apache.hadoop.hbase.HConstants.REPLICATION_SCOPE_LOCAL; @@ -62,6 +68,11 @@ public class Replication implements WALActionsListener, // Hosting server private Server server; + //Factories for lazy init. + private ReplicationZookeeperFactory replicationZookeeperFactory; + private ReplicationSourceManagerFactory replicationSourceManagerFactory; + private ReplicationSinkFactory replicationSinkFactory; + /** * Instantiate the replication management (if rep is enabled). * @param server Hosting server @@ -70,8 +81,17 @@ public class Replication implements WALActionsListener, * @param oldLogDir directory where logs are archived * @throws IOException */ - public Replication(final Server server, final FileSystem fs, - final Path logDir, final Path oldLogDir) throws IOException{ + @Inject + public Replication(@Assisted final Server server, + @Assisted final FileSystem fs, + @Assisted("logDir") final Path logDir, + @Assisted("oldLogDir") final Path oldLogDir, + ReplicationZookeeperFactory replicationZookeeperFactory, + ReplicationSourceManagerFactory replicationSourceManagerFactory, + ReplicationSinkFactory replicationSinkFactory) throws IOException{ + this.replicationZookeeperFactory = replicationZookeeperFactory; + this.replicationSourceManagerFactory = replicationSourceManagerFactory; + this.replicationSinkFactory = replicationSinkFactory; initialize(server, fs, logDir, oldLogDir); } @@ -88,12 +108,12 @@ public class Replication implements WALActionsListener, this.replication = isReplication(this.conf); if (replication) { try { - this.zkHelper = new ReplicationZookeeper(server, this.replicating); + this.zkHelper = replicationZookeeperFactory.create(server, this.replicating); } catch (KeeperException ke) { throw new IOException("Failed replication handler create " + "(replicating=" + this.replicating, ke); } - this.replicationManager = new ReplicationSourceManager(zkHelper, conf, + this.replicationManager = replicationSourceManagerFactory.create(zkHelper, this.server, fs, this.replicating, logDir, oldLogDir) ; } else { this.replicationManager = null; @@ -150,7 +170,7 @@ public class Replication implements WALActionsListener, public void startReplicationService() throws IOException { if (this.replication) { this.replicationManager.init(); - this.replicationSink = new ReplicationSink(this.conf, this.server); + this.replicationSink = replicationSinkFactory.create(); } } diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java index a359f78..3a03535 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java @@ -35,6 +35,7 @@ import org.apache.hadoop.hbase.regionserver.wal.WALEdit; import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationSinkMetrics; import org.apache.hadoop.hbase.util.Bytes; +import javax.inject.Inject; import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -74,12 +75,13 @@ public class ReplicationSink { * @param stopper boolean to tell this thread to stop * @throws IOException thrown when HDFS goes bad or bad file name */ - public ReplicationSink(Configuration conf, Stoppable stopper) + @Inject + public ReplicationSink(Configuration conf, ReplicationSinkMetrics metrics) throws IOException { this.conf = conf; this.pool = new HTablePool(this.conf, conf.getInt("replication.sink.htablepool.capacity", 10)); - this.metrics = new ReplicationSinkMetrics(); + this.metrics = metrics; } /** diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java index ddca9d1..8bdb781 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java @@ -37,6 +37,7 @@ import java.util.concurrent.PriorityBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience; @@ -51,6 +52,7 @@ import org.apache.hadoop.hbase.Stoppable; import org.apache.hadoop.hbase.client.AdminProtocol; import org.apache.hadoop.hbase.client.HConnection; import org.apache.hadoop.hbase.client.HConnectionManager; +import org.apache.hadoop.hbase.guice.ReplicationSourceMetricsFactory; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.regionserver.wal.HLog; import org.apache.hadoop.hbase.regionserver.wal.HLogKey; @@ -63,6 +65,8 @@ import org.apache.hadoop.hbase.zookeeper.ZKClusterId; import org.apache.hadoop.ipc.RemoteException; import org.apache.zookeeper.KeeperException; +import javax.inject.Inject; + /** * Class that handles the source of a replication stream. * Currently does not handle more than 1 slave @@ -141,24 +145,16 @@ public class ReplicationSource extends Thread // Metrics for this source private ReplicationSourceMetrics metrics; - /** - * Instantiation method used by region servers - * - * @param conf configuration to use - * @param fs file system to use - * @param manager replication manager to ping to - * @param stopper the atomic boolean to use to stop the regionserver - * @param replicating the atomic boolean that starts/stops replication - * @param peerClusterZnode the name of our znode - * @throws IOException - */ - public void init(final Configuration conf, - final FileSystem fs, - final ReplicationSourceManager manager, - final Stoppable stopper, - final AtomicBoolean replicating, - final String peerClusterZnode) - throws IOException { + public ReplicationSource() { } + + @Inject + public ReplicationSource(final Configuration conf, + final ReplicationSourceMetricsFactory replicationSourceMetricsFactory, + @Assisted final FileSystem fs, + @Assisted final ReplicationSourceManager manager, + @Assisted final Stoppable stopper, + @Assisted final AtomicBoolean replicating, + @Assisted final String peerClusterZnode) throws IOException { this.stopper = stopper; this.conf = conf; this.replicationQueueSizeCapacity = @@ -186,7 +182,7 @@ public class ReplicationSource extends Thread this.sleepForRetries = this.conf.getLong("replication.source.sleepforretries", 1000); this.fs = fs; - this.metrics = new ReplicationSourceMetrics(peerClusterZnode); + this.metrics = replicationSourceMetricsFactory.create(peerClusterZnode); try { this.clusterId = UUID.fromString(ZKClusterId.readClusterIdZNode(zkHelper diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java index ccafe1f..f4a660c 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java @@ -35,23 +35,6 @@ import org.apache.hadoop.hbase.Stoppable; public interface ReplicationSourceInterface { /** - * Initializer for the source - * @param conf the configuration to use - * @param fs the file system to use - * @param manager the manager to use - * @param stopper the stopper object for this region server - * @param replicating the status of the replication on this cluster - * @param peerClusterId the id of the peer cluster - * @throws IOException - */ - public void init(final Configuration conf, - final FileSystem fs, - final ReplicationSourceManager manager, - final Stoppable stopper, - final AtomicBoolean replicating, - final String peerClusterId) throws IOException; - - /** * Add a log to the list of logs to replicate * @param log path to the log to replicate */ diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java index 4a3ed90..7cb84bd 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java @@ -35,6 +35,7 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience; @@ -42,6 +43,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.Stoppable; +import org.apache.hadoop.hbase.guice.ReplicationSourceFactory; import org.apache.hadoop.hbase.replication.ReplicationZookeeper; import org.apache.hadoop.hbase.zookeeper.ZooKeeperListener; import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher; @@ -49,6 +51,8 @@ import org.apache.zookeeper.KeeperException; import com.google.common.util.concurrent.ThreadFactoryBuilder; +import javax.inject.Inject; + /** * This class is responsible to manage all the replication * sources. There are two classes of sources: @@ -90,6 +94,7 @@ public class ReplicationSourceManager { private final long sleepBeforeFailover; // Homemade executer service for replication private final ThreadPoolExecutor executor; + private ReplicationSourceFactory replicationSourceFactory; /** * Creates a replication manager and sets the watch on all the other @@ -102,13 +107,16 @@ public class ReplicationSourceManager { * @param logDir the directory that contains all hlog directories of live RSs * @param oldLogDir the directory where old logs are archived */ - public ReplicationSourceManager(final ReplicationZookeeper zkHelper, + @Inject + public ReplicationSourceManager(@Assisted final ReplicationZookeeper zkHelper, final Configuration conf, - final Stoppable stopper, - final FileSystem fs, - final AtomicBoolean replicating, - final Path logDir, - final Path oldLogDir) { + @Assisted final Stoppable stopper, + @Assisted final FileSystem fs, + @Assisted final AtomicBoolean replicating, + @Assisted("logDir") final Path logDir, + @Assisted("oldLogDir") final Path oldLogDir, + final ReplicationSourceFactory replicationSourceFactory) { + this.replicationSourceFactory = replicationSourceFactory; this.sources = new ArrayList(); this.replicating = replicating; this.zkHelper = zkHelper; @@ -201,7 +209,7 @@ public class ReplicationSourceManager { */ public ReplicationSourceInterface addSource(String id) throws IOException { ReplicationSourceInterface src = - getReplicationSource(this.conf, this.fs, this, stopper, replicating, id); + getReplicationSource(this.fs, this, stopper, replicating, id); synchronized (this.hlogsById) { this.sources.add(src); this.hlogsById.put(id, new TreeSet()); @@ -295,7 +303,6 @@ public class ReplicationSourceManager { /** * Factory method to create a replication source - * @param conf the configuration to use * @param fs the file system to use * @param manager the manager to use * @param stopper the stopper object for this region server @@ -305,25 +312,12 @@ public class ReplicationSourceManager { * @throws IOException */ public ReplicationSourceInterface getReplicationSource( - final Configuration conf, final FileSystem fs, final ReplicationSourceManager manager, final Stoppable stopper, final AtomicBoolean replicating, final String peerId) throws IOException { - ReplicationSourceInterface src; - try { - @SuppressWarnings("rawtypes") - Class c = Class.forName(conf.get("replication.replicationsource.implementation", - ReplicationSource.class.getCanonicalName())); - src = (ReplicationSourceInterface) c.newInstance(); - } catch (Exception e) { - LOG.warn("Passed replication source implementation throws errors, " + - "defaulting to ReplicationSource", e); - src = new ReplicationSource(); - - } - src.init(conf, fs, manager, stopper, replicating, peerId); + ReplicationSourceInterface src = this.replicationSourceFactory.create(fs, manager, stopper, replicating, peerId); return src; } @@ -584,8 +578,8 @@ public class ReplicationSourceManager { for (Map.Entry> entry : newQueues.entrySet()) { String peerId = entry.getKey(); try { - ReplicationSourceInterface src = getReplicationSource(conf, - fs, ReplicationSourceManager.this, stopper, replicating, peerId); + ReplicationSourceInterface src = + getReplicationSource(fs, ReplicationSourceManager.this, stopper, replicating, peerId); if (!zkHelper.getPeerClusters().containsKey(src.getPeerClusterId())) { src.terminate("Recovered queue doesn't belong to any current peer"); break; diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationSinkMetrics.java hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationSinkMetrics.java index 7bab049..f56593f 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationSinkMetrics.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationSinkMetrics.java @@ -19,8 +19,8 @@ package org.apache.hadoop.hbase.replication.regionserver.metrics; import org.apache.hadoop.classification.InterfaceAudience; -import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSource; -import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSourceFactory; + +import javax.inject.Inject; /** * This class is for maintaining the various replication statistics for a sink and publishing them @@ -35,8 +35,10 @@ public class ReplicationSinkMetrics { private ReplicationMetricsSource rms; - public ReplicationSinkMetrics() { - rms = ReplicationMetricsSourceFactory.getInstance(); + @Inject + public ReplicationSinkMetrics(ReplicationMetricsSource rms) { + this.rms = rms; + this.applyBatch(101); } /** diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationSourceMetrics.java hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationSourceMetrics.java index 35e3f55..13db1e3 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationSourceMetrics.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/metrics/ReplicationSourceMetrics.java @@ -18,10 +18,13 @@ package org.apache.hadoop.hbase.replication.regionserver.metrics; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience; +import javax.inject.Inject; + /** * This class is for maintaining the various replication statistics for a source and publishing them * through the metrics interfaces. @@ -56,7 +59,8 @@ public class ReplicationSourceMetrics { * * @param id Name of the source this class is monitoring */ - public ReplicationSourceMetrics(String id) { + @Inject + public ReplicationSourceMetrics(@Assisted String id, ReplicationMetricsSource rms) { this.id = id; sizeOfLogQueKey = "source." + id + ".sizeOfLogQueue"; @@ -65,7 +69,7 @@ public class ReplicationSourceMetrics { logEditsFilteredKey = "source." + id + ".logEditsFiltered"; shippedBatchesKey = "source." + this.id + ".shippedBatches"; shippedOpsKey = "source." + this.id + ".shippedOps"; - rms = ReplicationMetricsSourceFactory.getInstance(); + this.rms = rms; } /** diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/util/JVMClusterUtil.java hbase-server/src/main/java/org/apache/hadoop/hbase/util/JVMClusterUtil.java index 598e851..b5bcac4 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/util/JVMClusterUtil.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/util/JVMClusterUtil.java @@ -20,7 +20,6 @@ package org.apache.hadoop.hbase.util; import java.io.IOException; -import java.lang.reflect.InvocationTargetException; import java.util.List; import org.apache.commons.logging.Log; @@ -32,6 +31,8 @@ import org.apache.hadoop.hbase.master.HMaster; import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.regionserver.ShutdownHook; +import javax.inject.Provider; + /** * Utility used running a cluster all in the one JVM. */ @@ -71,29 +72,16 @@ public class JVMClusterUtil { /** * Creates a {@link RegionServerThread}. * Call 'start' on the returned thread to make it run. - * @param c Configuration to use. - * @param hrsc Class to create. * @param index Used distinguishing the object returned. * @throws IOException * @return Region server added. */ public static JVMClusterUtil.RegionServerThread createRegionServerThread( - final Configuration c, final Class hrsc, + final Provider hrsf, + final Configuration conf, final int index) throws IOException { - HRegionServer server; - try { - server = hrsc.getConstructor(Configuration.class).newInstance(c); - } catch (InvocationTargetException ite) { - Throwable target = ite.getTargetException(); - throw new RuntimeException("Failed construction of RegionServer: " + - hrsc.toString() + ((target.getCause() != null)? - target.getCause().getMessage(): ""), target); - } catch (Exception e) { - IOException ioe = new IOException(); - ioe.initCause(e); - throw ioe; - } + HRegionServer server = hrsf.get(); return new JVMClusterUtil.RegionServerThread(server, index); } @@ -118,29 +106,16 @@ public class JVMClusterUtil { /** * Creates a {@link MasterThread}. * Call 'start' on the returned thread to make it run. - * @param c Configuration to use. - * @param hmc Class to create. * @param index Used distinguishing the object returned. * @throws IOException * @return Master added. */ public static JVMClusterUtil.MasterThread createMasterThread( - final Configuration c, final Class hmc, + final Provider hmf, + final Configuration conf, final int index) throws IOException { - HMaster server; - try { - server = hmc.getConstructor(Configuration.class).newInstance(c); - } catch (InvocationTargetException ite) { - Throwable target = ite.getTargetException(); - throw new RuntimeException("Failed construction of Master: " + - hmc.toString() + ((target.getCause() != null)? - target.getCause().getMessage(): ""), target); - } catch (Exception e) { - IOException ioe = new IOException(); - ioe.initCause(e); - throw ioe; - } + HMaster server = hmf.get(); return new JVMClusterUtil.MasterThread(server, index); } diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/util/Sleeper.java hbase-server/src/main/java/org/apache/hadoop/hbase/util/Sleeper.java index c8b657a..4624a27 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/util/Sleeper.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/util/Sleeper.java @@ -19,12 +19,15 @@ */ package org.apache.hadoop.hbase.util; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; import org.apache.hadoop.hbase.Stoppable; +import javax.inject.Inject; + /** * Sleeper for current thread. * Sleeps for passed period. Also checks passed boolean and if interrupted, @@ -47,7 +50,8 @@ public class Sleeper { * @param stopper When {@link Stoppable#isStopped()} is true, this thread will * cleanup and exit cleanly. */ - public Sleeper(final int sleep, final Stoppable stopper) { + @Inject + public Sleeper(@Assisted final int sleep, @Assisted final Stoppable stopper) { this.period = sleep; this.stopper = stopper; } diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java index 33bc1d0..ee03387 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java @@ -26,6 +26,7 @@ import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience; @@ -41,6 +42,8 @@ import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.data.ACL; +import javax.inject.Inject; + /** * Acts as the single ZooKeeper Watcher. One instance of this is instantiated * for each Master, RegionServer, and client process. @@ -132,8 +135,9 @@ public class ZooKeeperWatcher implements Watcher, Abortable, Closeable { * @throws IOException * @throws ZooKeeperConnectionException */ - public ZooKeeperWatcher(Configuration conf, String descriptor, - Abortable abortable, boolean canCreateBaseZNode) + @Inject + public ZooKeeperWatcher(Configuration conf, @Assisted String descriptor, + @Assisted Abortable abortable, @Assisted boolean canCreateBaseZNode) throws IOException, ZooKeeperConnectionException { this.conf = conf; // Capture a stack trace now. Will print it out later if problem so we can diff --git hbase-server/src/main/resources/hbase-default.xml hbase-server/src/main/resources/hbase-default.xml index a42d94b..af33909 100644 --- hbase-server/src/main/resources/hbase-default.xml +++ hbase-server/src/main/resources/hbase-default.xml @@ -895,4 +895,9 @@ default log cleaners in the list as they will be overwritten in hbase-site.xml. + + + hbase.replication + true + diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java index 0259470..9ebc9fa 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java @@ -43,6 +43,9 @@ import java.util.Random; import java.util.Set; import java.util.UUID; +import com.google.inject.Guice; +import com.google.inject.Injector; +import com.google.inject.Module; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.impl.Jdk14Logger; @@ -62,6 +65,9 @@ import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.fs.HFileSystem; +import org.apache.hadoop.hbase.guice.HBaseGuice; +import org.apache.hadoop.hbase.guice.HBaseGuiceTestUtil; +import org.apache.hadoop.hbase.guice.MiniHBaseClusterFactory; import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding; import org.apache.hadoop.hbase.io.hfile.ChecksumUtil; import org.apache.hadoop.hbase.io.hfile.Compression; @@ -186,6 +192,8 @@ public class HBaseTestingUtility { Compression.Algorithm.NONE, Compression.Algorithm.GZ }; + private Injector injector; + /** * Create all combinations of Bloom filters and compression algorithms for * testing. @@ -210,6 +218,7 @@ public class HBaseTestingUtility { public HBaseTestingUtility(Configuration conf) { this.conf = conf; + this.injector = HBaseGuiceTestUtil.createDefaultInector(conf); // a hbase checksum verification failure will cause unit tests to fail ChecksumUtil.generateExceptionForChecksumFailureForTest(true); @@ -230,6 +239,14 @@ public class HBaseTestingUtility { return this.conf; } + public Injector getInjector() { + return injector; + } + + public void setInjector(Injector injector) { + this.injector = injector; + } + /** * @return Where to write test data on local filesystem; usually * {@link #DEFAULT_BASE_TEST_DIRECTORY} @@ -647,7 +664,7 @@ public class HBaseTestingUtility { * @throws InterruptedException * @see {@link #startMiniCluster()} */ - public MiniHBaseCluster startMiniHBaseCluster(final int numMasters, + public MiniHBaseCluster startMiniHBaseCluster( final int numMasters, final int numSlaves) throws IOException, InterruptedException { // Now do the mini hbase cluster. Set the hbase.rootdir in config. @@ -659,7 +676,7 @@ public class HBaseTestingUtility { conf.setInt("hbase.master.wait.on.regionservers.maxtostart", numSlaves); Configuration c = new Configuration(this.conf); - this.hbaseCluster = new MiniHBaseCluster(c, numMasters, numSlaves); + this.hbaseCluster = injector.getInstance(MiniHBaseClusterFactory.class).create(numMasters, numSlaves); // Don't leave here till we've done a successful scan of the .META. HTable t = new HTable(c, HConstants.META_TABLE_NAME); ResultScanner s = t.getScanner(new Scan()); @@ -681,7 +698,7 @@ public class HBaseTestingUtility { * @throws IOException */ public void restartHBaseCluster(int servers) throws IOException, InterruptedException { - this.hbaseCluster = new MiniHBaseCluster(this.conf, servers); + this.hbaseCluster = injector.getInstance(MiniHBaseClusterFactory.class).create(1, servers); // Don't leave here till we've done a successful scan of the .META. HTable t = new HTable(new Configuration(this.conf), HConstants.META_TABLE_NAME); ResultScanner s = t.getScanner(new Scan()); diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/MiniHBaseCluster.java hbase-server/src/test/java/org/apache/hadoop/hbase/MiniHBaseCluster.java index c7442ae..2bb1006 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/MiniHBaseCluster.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/MiniHBaseCluster.java @@ -24,6 +24,8 @@ import java.security.PrivilegedAction; import java.util.ArrayList; import java.util.List; +import com.google.inject.Injector; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.classification.InterfaceAudience; @@ -31,17 +33,25 @@ import org.apache.hadoop.classification.InterfaceStability; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.hbase.client.HConnectionManager; +import org.apache.hadoop.hbase.guice.HBaseGuice; +import org.apache.hadoop.hbase.guice.LocalHBaseClusterFactory; +import org.apache.hadoop.hbase.guice.ReplicationFactory; +import org.apache.hadoop.hbase.guice.RpcServerFactory; +import org.apache.hadoop.hbase.guice.SleeperFactory; +import org.apache.hadoop.hbase.guice.ZooKeeperWatcherFactory; import org.apache.hadoop.hbase.master.HMaster; import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupResponse; import org.apache.hadoop.hbase.regionserver.HRegion; import org.apache.hadoop.hbase.regionserver.HRegionServer; +import org.apache.hadoop.hbase.regionserver.RegionServerAccounting; import org.apache.hadoop.hbase.security.User; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.JVMClusterUtil.MasterThread; import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread; import org.apache.hadoop.hbase.util.JVMClusterUtil; import org.apache.hadoop.hbase.util.Threads; -import org.apache.hadoop.io.MapWritable; + +import javax.inject.Inject; /** * This class creates a single process HBase cluster. @@ -54,31 +64,25 @@ import org.apache.hadoop.io.MapWritable; public class MiniHBaseCluster { static final Log LOG = LogFactory.getLog(MiniHBaseCluster.class.getName()); private Configuration conf; + private LocalHBaseClusterFactory localHBaseClusterFactory; public LocalHBaseCluster hbaseCluster; private static int index; /** * Start a MiniHBaseCluster. * @param conf Configuration to be used for cluster - * @param numRegionServers initial number of region servers to start. - * @throws IOException - */ - public MiniHBaseCluster(Configuration conf, int numRegionServers) - throws IOException, InterruptedException { - this(conf, 1, numRegionServers); - } - - /** - * Start a MiniHBaseCluster. - * @param conf Configuration to be used for cluster * @param numMasters initial number of masters to start. * @param numRegionServers initial number of region servers to start. * @throws IOException */ - public MiniHBaseCluster(Configuration conf, int numMasters, - int numRegionServers) + @Inject + public MiniHBaseCluster(Configuration conf, + LocalHBaseClusterFactory localHBaseClusterFactory, + @Assisted("numMasters") int numMasters, + @Assisted("numRegionServers") int numRegionServers) throws IOException, InterruptedException { this.conf = conf; + this.localHBaseClusterFactory = localHBaseClusterFactory; conf.set(HConstants.MASTER_PORT, "0"); init(numMasters, numRegionServers); } @@ -98,19 +102,36 @@ public class MiniHBaseCluster { private User user = null; public static boolean TEST_SKIP_CLOSE = false; - public MiniHBaseClusterRegionServer(Configuration conf) + /** + * Starts a HRegionServer at the default location + * + * @param conf + * @throws java.io.IOException + * @throws InterruptedException + */ + public MiniHBaseClusterRegionServer(Configuration conf, + SleeperFactory sleeperFactory, + RpcServerFactory rpcServerFactory, + ReplicationFactory replicationFactory, + RegionServerAccounting rsAccounting, + ZooKeeperWatcherFactory zooKeeperWatcherFactory) throws IOException, InterruptedException { - super(conf); - this.user = User.getCurrent(); + super(conf, + sleeperFactory, + rpcServerFactory, + replicationFactory, + rsAccounting, + zooKeeperWatcherFactory); + user = User.getCurrent(); } /* - * @param c - * @param currentfs We return this if we did not make a new one. - * @param uniqueName Same name used to help identify the created fs. - * @return A new fs instance if we are up on DistributeFileSystem. - * @throws IOException - */ + * @param c + * @param currentfs We return this if we did not make a new one. + * @param uniqueName Same name used to help identify the created fs. + * @return A new fs instance if we are up on DistributeFileSystem. + * @throws IOException + */ @Override protected void handleReportForDutyResponse( @@ -190,15 +211,14 @@ public class MiniHBaseCluster { throws IOException, InterruptedException { try { // start up a LocalHBaseCluster - hbaseCluster = new LocalHBaseCluster(conf, nMasterNodes, 0, - HMaster.class, MiniHBaseCluster.MiniHBaseClusterRegionServer.class); + hbaseCluster = localHBaseClusterFactory.create(nMasterNodes,0); // manually add the regionservers as other users for (int i=0; i modules = HBaseGuice.makeLocalHBaseModules(conf); + modules.put("LocalHBaseCluster", new TestLocalHBaseClusterModule()); + + Injector injector = Guice.createInjector(modules.values()); + LocalHBaseCluster cluster = injector.getInstance(LocalHBaseClusterFactory.class).create(1,1); + // Can we cast back to our master class? try { ((MyHMaster)cluster.getMaster(0)).setZKCluster(zkCluster); @@ -89,7 +111,8 @@ public class TestLocalHBaseCluster { public static class MyHMaster extends HMaster { private MiniZooKeeperCluster zkcluster = null; - public MyHMaster(Configuration conf) throws IOException, KeeperException, + @Inject + public MyHMaster(@Assisted Configuration conf) throws IOException, KeeperException, InterruptedException { super(conf); } @@ -116,13 +139,43 @@ public class TestLocalHBaseCluster { */ public static class MyHRegionServer extends HRegionServer { - public MyHRegionServer(Configuration conf) throws IOException, - InterruptedException { - super(conf); + /** + * Starts a HRegionServer at the default location + * + * @param conf + * @throws java.io.IOException + * @throws InterruptedException + */ + public MyHRegionServer(Configuration conf, + SleeperFactory sleeperFactory, + RpcServerFactory rpcServerFactory, + ReplicationFactory replicationFactory, + RegionServerAccounting rsAccounting, + ZooKeeperWatcherFactory zooKeeperWatcherFactory) + throws IOException, InterruptedException { + super(conf, + sleeperFactory, + rpcServerFactory, + replicationFactory, + rsAccounting, + zooKeeperWatcherFactory); } public int echo(int val) { return val; } } + + public static class TestLocalHBaseClusterModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(LocalHBaseCluster.class, LocalHBaseCluster.class) + .build(LocalHBaseClusterFactory.class)); + + bind(HRegionServer.class).to(MyHRegionServer.class); + bind(HMaster.class).to(MyHMaster.class); + } + } } diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java index 6dabc27..78292d2 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java @@ -25,6 +25,9 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.*; +import org.apache.hadoop.hbase.guice.HBaseGuiceTestUtil; +import org.apache.hadoop.hbase.guice.ReplicationFactory; +import org.apache.hadoop.hbase.guice.ReplicationSourceManagerFactory; import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager; import org.junit.BeforeClass; import org.junit.Test; @@ -66,9 +69,7 @@ public class TestReplicationAdmin { HConstants.HREGION_OLDLOGDIR_NAME); Path logDir = new Path(TEST_UTIL.getDataTestDir(), HConstants.HREGION_LOGDIR_NAME); - manager = new ReplicationSourceManager(admin.getReplicationZk(), conf, - // The following stopper never stops so that we can respond - // to zk notification + manager = HBaseGuiceTestUtil.createDefaultInector(conf).getInstance(ReplicationSourceManagerFactory.class).create(admin.getReplicationZk(), new Stoppable() { @Override public void stop(String why) {} diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/guice/HBaseGuiceTestUtil.java hbase-server/src/test/java/org/apache/hadoop/hbase/guice/HBaseGuiceTestUtil.java new file mode 100644 index 0000000..0c071b0 --- /dev/null +++ hbase-server/src/test/java/org/apache/hadoop/hbase/guice/HBaseGuiceTestUtil.java @@ -0,0 +1,45 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.Guice; +import com.google.inject.Injector; +import com.google.inject.Module; +import org.apache.hadoop.conf.Configuration; + +import java.util.Collection; +import java.util.Map; + +/** + * + */ +public class HBaseGuiceTestUtil { + public static Injector createDefaultInector(Configuration conf) { + Collection modules = makeDefaultTestModules(conf).values(); + + return Guice.createInjector(modules); + } + + public static Map makeDefaultTestModules(Configuration conf) { + Map modules = HBaseGuice.makeLocalHBaseModules(conf); + modules.put("MiniHBaseCluster", new MiniHBaseClusterModule()); + return modules; + } + +} diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/guice/MiniHBaseClusterFactory.java hbase-server/src/test/java/org/apache/hadoop/hbase/guice/MiniHBaseClusterFactory.java new file mode 100644 index 0000000..707610c --- /dev/null +++ hbase-server/src/test/java/org/apache/hadoop/hbase/guice/MiniHBaseClusterFactory.java @@ -0,0 +1,32 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.assistedinject.Assisted; +import org.apache.hadoop.hbase.MiniHBaseCluster; + +/** + * + */ +public interface MiniHBaseClusterFactory { + + public MiniHBaseCluster create(@Assisted("numMasters") int numMasters, + @Assisted("numRegionServers") int numRegionServers); + +} diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/guice/MiniHBaseClusterModule.java hbase-server/src/test/java/org/apache/hadoop/hbase/guice/MiniHBaseClusterModule.java new file mode 100644 index 0000000..6a094cb --- /dev/null +++ hbase-server/src/test/java/org/apache/hadoop/hbase/guice/MiniHBaseClusterModule.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.hadoop.hbase.guice; + +import com.google.inject.AbstractModule; +import com.google.inject.assistedinject.FactoryModuleBuilder; +import org.apache.hadoop.hbase.MiniHBaseCluster; + +/** + * + */ +public class MiniHBaseClusterModule extends AbstractModule { + + @Override + protected void configure() { + install(new FactoryModuleBuilder() + .implement(MiniHBaseCluster.class, MiniHBaseCluster.class) + .build(MiniHBaseClusterFactory.class)); + } +} diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/OOMERegionServer.java hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/OOMERegionServer.java index fa177ae..0353770 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/OOMERegionServer.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/OOMERegionServer.java @@ -23,8 +23,12 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; -import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.guice.ReplicationFactory; +import org.apache.hadoop.hbase.guice.RpcServerFactory; +import org.apache.hadoop.hbase.guice.SleeperFactory; +import org.apache.hadoop.hbase.guice.ZooKeeperWatcherFactory; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.RequestConverter; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateRequest; @@ -41,8 +45,26 @@ import com.google.protobuf.ServiceException; public class OOMERegionServer extends HRegionServer { private List retainer = new ArrayList(); - public OOMERegionServer(HBaseConfiguration conf) throws IOException, InterruptedException { - super(conf); + /** + * Starts a HRegionServer at the default location + * + * @param conf + * @throws java.io.IOException + * @throws InterruptedException + */ + public OOMERegionServer(Configuration conf, + SleeperFactory sleeperFactory, + RpcServerFactory rpcServerFactory, + ReplicationFactory replicationFactory, + RegionServerAccounting rsAccounting, + ZooKeeperWatcherFactory zooKeeperWatcherFactory) + throws IOException, InterruptedException { + super(conf, + sleeperFactory, + rpcServerFactory, + replicationFactory, + rsAccounting, + zooKeeperWatcherFactory); } public void put(byte [] regionName, Put put) diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRpcMetrics.java hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRpcMetrics.java index 0a90f0f..d288420 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRpcMetrics.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRpcMetrics.java @@ -25,11 +25,17 @@ import java.util.HashMap; import java.util.Map; import java.util.Random; +import com.google.inject.assistedinject.Assisted; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.hadoop.hbase.MediumTests; +import org.apache.hadoop.hbase.guice.HBaseGuiceTestUtil; +import org.apache.hadoop.hbase.guice.ReplicationFactory; +import org.apache.hadoop.hbase.guice.RpcServerFactory; +import org.apache.hadoop.hbase.guice.SleeperFactory; +import org.apache.hadoop.hbase.guice.ZooKeeperWatcherFactory; import org.apache.hadoop.hbase.ipc.HBaseRpcMetrics; import org.apache.hadoop.metrics.ContextFactory; import org.apache.hadoop.metrics.MetricsContext; @@ -57,11 +63,26 @@ public class TestRpcMetrics { */ public static class TestRegionServer extends HRegionServer { - public TestRegionServer(Configuration conf) + /** + * Starts a HRegionServer at the default location + * + * @param conf + * @throws java.io.IOException + * @throws InterruptedException + */ + public TestRegionServer(Configuration conf, + SleeperFactory sleeperFactory, + RpcServerFactory rpcServerFactory, + ReplicationFactory replicationFactory, + RegionServerAccounting rsAccounting, + ZooKeeperWatcherFactory zooKeeperWatcherFactory) throws IOException, InterruptedException { - super(conf); - - // register custom metrics interface + super(conf, + sleeperFactory, + rpcServerFactory, + replicationFactory, + rsAccounting, + zooKeeperWatcherFactory); getRpcMetrics().createMetrics(new Class[]{TestMetrics.class}, true); } @@ -120,7 +141,8 @@ public class TestRpcMetrics { @Test public void testCustomMetrics() throws Exception { TEST_UTIL.getConfiguration().setInt("hbase.regionserver.port", 0); - TestRegionServer rs = new TestRegionServer(TEST_UTIL.getConfiguration()); + TestRegionServer rs = HBaseGuiceTestUtil.createDefaultInector(TEST_UTIL.getConfiguration()) + .getInstance(TestRegionServer.class); rs.incTest(5); // wait for metrics context update diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSink.java hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSink.java index 18eb530..3af5eb7 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSink.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSink.java @@ -36,6 +36,8 @@ import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.regionserver.wal.HLog; import org.apache.hadoop.hbase.regionserver.wal.HLogKey; import org.apache.hadoop.hbase.regionserver.wal.WALEdit; +import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationMetricsSourceImpl; +import org.apache.hadoop.hbase.replication.regionserver.metrics.ReplicationSinkMetrics; import org.apache.hadoop.hbase.util.Bytes; import org.junit.AfterClass; import org.junit.Before; @@ -89,7 +91,7 @@ public class TestReplicationSink { TEST_UTIL.getConfiguration().setBoolean(HConstants.REPLICATION_ENABLE_KEY, true); TEST_UTIL.startMiniCluster(3); SINK = - new ReplicationSink(new Configuration(TEST_UTIL.getConfiguration()), STOPPABLE); + new ReplicationSink(new Configuration(TEST_UTIL.getConfiguration()), new ReplicationSinkMetrics(new ReplicationMetricsSourceImpl())); table1 = TEST_UTIL.createTable(TABLE_NAME1, FAM_NAME1); table2 = TEST_UTIL.createTable(TABLE_NAME2, FAM_NAME2); } diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java index 5828154..6cb3d65 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java @@ -41,6 +41,8 @@ import org.apache.hadoop.hbase.MediumTests; import org.apache.hadoop.hbase.Server; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.catalog.CatalogTracker; +import org.apache.hadoop.hbase.guice.HBaseGuiceTestUtil; +import org.apache.hadoop.hbase.guice.ReplicationFactory; import org.apache.hadoop.hbase.regionserver.wal.HLog; import org.apache.hadoop.hbase.regionserver.wal.HLogKey; import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener; @@ -116,7 +118,8 @@ public class TestReplicationSourceManager { ZKUtil.createWithParents(zkw, "/hbase/replication/state"); ZKUtil.setData(zkw, "/hbase/replication/state", Bytes.toBytes("true")); - replication = new Replication(new DummyServer(), fs, logDir, oldLogDir); + + replication = HBaseGuiceTestUtil.createDefaultInector(conf).getInstance(ReplicationFactory.class).create(new DummyServer(), fs, logDir, oldLogDir); manager = replication.getReplicationManager(); fs = FileSystem.get(conf); oldLogDir = new Path(utility.getDataTestDir(), diff --git pom.xml pom.xml index 7a8a643..6c5193f 100644 --- pom.xml +++ pom.xml @@ -829,6 +829,7 @@ 2.3.1 1.3.1 3.5.0.Final-SNAPSHOT + 3.0 2.3 1.6 @@ -903,6 +904,16 @@ + com.google.inject + guice + ${guice.version} + + + com.google.inject.extensions + guice-assistedinject + ${guice.version} + + io.netty netty ${netty.version}