http://git-wip-us.apache.org/repos/asf/hbase-site/blob/738e976e/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegionServer.html ---------------------------------------------------------------------- diff --git a/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegionServer.html b/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegionServer.html index e074a8c..8cc5add 100644 --- a/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegionServer.html +++ b/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HRegionServer.html @@ -722,172 +722,172 @@ <span class="sourceLineNo">714</span> "hbase.regionserver.kerberos.principal", host);<a name="line.714"></a> <span class="sourceLineNo">715</span> }<a name="line.715"></a> <span class="sourceLineNo">716</span><a name="line.716"></a> -<span class="sourceLineNo">717</span> protected void waitForMasterActive() {<a name="line.717"></a> -<span class="sourceLineNo">718</span> }<a name="line.718"></a> -<span class="sourceLineNo">719</span><a name="line.719"></a> -<span class="sourceLineNo">720</span> protected String getProcessName() {<a name="line.720"></a> -<span class="sourceLineNo">721</span> return REGIONSERVER;<a name="line.721"></a> -<span class="sourceLineNo">722</span> }<a name="line.722"></a> +<span class="sourceLineNo">717</span><a name="line.717"></a> +<span class="sourceLineNo">718</span> /**<a name="line.718"></a> +<span class="sourceLineNo">719</span> * Wait for an active Master.<a name="line.719"></a> +<span class="sourceLineNo">720</span> * See override in Master superclass for how it is used.<a name="line.720"></a> +<span class="sourceLineNo">721</span> */<a name="line.721"></a> +<span class="sourceLineNo">722</span> protected void waitForMasterActive() {}<a name="line.722"></a> <span class="sourceLineNo">723</span><a name="line.723"></a> -<span class="sourceLineNo">724</span> protected boolean canCreateBaseZNode() {<a name="line.724"></a> -<span class="sourceLineNo">725</span> return this.masterless;<a name="line.725"></a> +<span class="sourceLineNo">724</span> protected String getProcessName() {<a name="line.724"></a> +<span class="sourceLineNo">725</span> return REGIONSERVER;<a name="line.725"></a> <span class="sourceLineNo">726</span> }<a name="line.726"></a> <span class="sourceLineNo">727</span><a name="line.727"></a> -<span class="sourceLineNo">728</span> protected boolean canUpdateTableDescriptor() {<a name="line.728"></a> -<span class="sourceLineNo">729</span> return false;<a name="line.729"></a> +<span class="sourceLineNo">728</span> protected boolean canCreateBaseZNode() {<a name="line.728"></a> +<span class="sourceLineNo">729</span> return this.masterless;<a name="line.729"></a> <span class="sourceLineNo">730</span> }<a name="line.730"></a> <span class="sourceLineNo">731</span><a name="line.731"></a> -<span class="sourceLineNo">732</span> protected RSRpcServices createRpcServices() throws IOException {<a name="line.732"></a> -<span class="sourceLineNo">733</span> return new RSRpcServices(this);<a name="line.733"></a> +<span class="sourceLineNo">732</span> protected boolean canUpdateTableDescriptor() {<a name="line.732"></a> +<span class="sourceLineNo">733</span> return false;<a name="line.733"></a> <span class="sourceLineNo">734</span> }<a name="line.734"></a> <span class="sourceLineNo">735</span><a name="line.735"></a> -<span class="sourceLineNo">736</span> protected void configureInfoServer() {<a name="line.736"></a> -<span class="sourceLineNo">737</span> infoServer.addServlet("rs-status", "/rs-status", RSStatusServlet.class);<a name="line.737"></a> -<span class="sourceLineNo">738</span> infoServer.setAttribute(REGIONSERVER, this);<a name="line.738"></a> -<span class="sourceLineNo">739</span> }<a name="line.739"></a> -<span class="sourceLineNo">740</span><a name="line.740"></a> -<span class="sourceLineNo">741</span> protected Class<? extends HttpServlet> getDumpServlet() {<a name="line.741"></a> -<span class="sourceLineNo">742</span> return RSDumpServlet.class;<a name="line.742"></a> +<span class="sourceLineNo">736</span> protected RSRpcServices createRpcServices() throws IOException {<a name="line.736"></a> +<span class="sourceLineNo">737</span> return new RSRpcServices(this);<a name="line.737"></a> +<span class="sourceLineNo">738</span> }<a name="line.738"></a> +<span class="sourceLineNo">739</span><a name="line.739"></a> +<span class="sourceLineNo">740</span> protected void configureInfoServer() {<a name="line.740"></a> +<span class="sourceLineNo">741</span> infoServer.addServlet("rs-status", "/rs-status", RSStatusServlet.class);<a name="line.741"></a> +<span class="sourceLineNo">742</span> infoServer.setAttribute(REGIONSERVER, this);<a name="line.742"></a> <span class="sourceLineNo">743</span> }<a name="line.743"></a> <span class="sourceLineNo">744</span><a name="line.744"></a> -<span class="sourceLineNo">745</span> @Override<a name="line.745"></a> -<span class="sourceLineNo">746</span> public boolean registerService(com.google.protobuf.Service instance) {<a name="line.746"></a> -<span class="sourceLineNo">747</span> /*<a name="line.747"></a> -<span class="sourceLineNo">748</span> * No stacking of instances is allowed for a single executorService name<a name="line.748"></a> -<span class="sourceLineNo">749</span> */<a name="line.749"></a> -<span class="sourceLineNo">750</span> com.google.protobuf.Descriptors.ServiceDescriptor serviceDesc =<a name="line.750"></a> -<span class="sourceLineNo">751</span> instance.getDescriptorForType();<a name="line.751"></a> -<span class="sourceLineNo">752</span> String serviceName = CoprocessorRpcUtils.getServiceName(serviceDesc);<a name="line.752"></a> -<span class="sourceLineNo">753</span> if (coprocessorServiceHandlers.containsKey(serviceName)) {<a name="line.753"></a> -<span class="sourceLineNo">754</span> LOG.error("Coprocessor executorService " + serviceName<a name="line.754"></a> -<span class="sourceLineNo">755</span> + " already registered, rejecting request from " + instance);<a name="line.755"></a> -<span class="sourceLineNo">756</span> return false;<a name="line.756"></a> -<span class="sourceLineNo">757</span> }<a name="line.757"></a> -<span class="sourceLineNo">758</span><a name="line.758"></a> -<span class="sourceLineNo">759</span> coprocessorServiceHandlers.put(serviceName, instance);<a name="line.759"></a> -<span class="sourceLineNo">760</span> if (LOG.isDebugEnabled()) {<a name="line.760"></a> -<span class="sourceLineNo">761</span> LOG.debug("Registered regionserver coprocessor executorService: executorService=" + serviceName);<a name="line.761"></a> -<span class="sourceLineNo">762</span> }<a name="line.762"></a> -<span class="sourceLineNo">763</span> return true;<a name="line.763"></a> -<span class="sourceLineNo">764</span> }<a name="line.764"></a> -<span class="sourceLineNo">765</span><a name="line.765"></a> -<span class="sourceLineNo">766</span> /**<a name="line.766"></a> -<span class="sourceLineNo">767</span> * Create a 'smarter' Connection, one that is capable of by-passing RPC if the request is to<a name="line.767"></a> -<span class="sourceLineNo">768</span> * the local server; i.e. a short-circuit Connection. Safe to use going to local or remote<a name="line.768"></a> -<span class="sourceLineNo">769</span> * server. Create this instance in a method can be intercepted and mocked in tests.<a name="line.769"></a> -<span class="sourceLineNo">770</span> * @throws IOException<a name="line.770"></a> -<span class="sourceLineNo">771</span> */<a name="line.771"></a> -<span class="sourceLineNo">772</span> @VisibleForTesting<a name="line.772"></a> -<span class="sourceLineNo">773</span> protected ClusterConnection createClusterConnection() throws IOException {<a name="line.773"></a> -<span class="sourceLineNo">774</span> Configuration conf = this.conf;<a name="line.774"></a> -<span class="sourceLineNo">775</span> if (conf.get(HConstants.CLIENT_ZOOKEEPER_QUORUM) != null) {<a name="line.775"></a> -<span class="sourceLineNo">776</span> // Use server ZK cluster for server-issued connections, so we clone<a name="line.776"></a> -<span class="sourceLineNo">777</span> // the conf and unset the client ZK related properties<a name="line.777"></a> -<span class="sourceLineNo">778</span> conf = new Configuration(this.conf);<a name="line.778"></a> -<span class="sourceLineNo">779</span> conf.unset(HConstants.CLIENT_ZOOKEEPER_QUORUM);<a name="line.779"></a> -<span class="sourceLineNo">780</span> }<a name="line.780"></a> -<span class="sourceLineNo">781</span> // Create a cluster connection that when appropriate, can short-circuit and go directly to the<a name="line.781"></a> -<span class="sourceLineNo">782</span> // local server if the request is to the local server bypassing RPC. Can be used for both local<a name="line.782"></a> -<span class="sourceLineNo">783</span> // and remote invocations.<a name="line.783"></a> -<span class="sourceLineNo">784</span> return ConnectionUtils.createShortCircuitConnection(conf, null, userProvider.getCurrent(),<a name="line.784"></a> -<span class="sourceLineNo">785</span> serverName, rpcServices, rpcServices);<a name="line.785"></a> -<span class="sourceLineNo">786</span> }<a name="line.786"></a> -<span class="sourceLineNo">787</span><a name="line.787"></a> -<span class="sourceLineNo">788</span> /**<a name="line.788"></a> -<span class="sourceLineNo">789</span> * Run test on configured codecs to make sure supporting libs are in place.<a name="line.789"></a> -<span class="sourceLineNo">790</span> * @param c<a name="line.790"></a> -<span class="sourceLineNo">791</span> * @throws IOException<a name="line.791"></a> -<span class="sourceLineNo">792</span> */<a name="line.792"></a> -<span class="sourceLineNo">793</span> private static void checkCodecs(final Configuration c) throws IOException {<a name="line.793"></a> -<span class="sourceLineNo">794</span> // check to see if the codec list is available:<a name="line.794"></a> -<span class="sourceLineNo">795</span> String [] codecs = c.getStrings("hbase.regionserver.codecs", (String[])null);<a name="line.795"></a> -<span class="sourceLineNo">796</span> if (codecs == null) return;<a name="line.796"></a> -<span class="sourceLineNo">797</span> for (String codec : codecs) {<a name="line.797"></a> -<span class="sourceLineNo">798</span> if (!CompressionTest.testCompression(codec)) {<a name="line.798"></a> -<span class="sourceLineNo">799</span> throw new IOException("Compression codec " + codec +<a name="line.799"></a> -<span class="sourceLineNo">800</span> " not supported, aborting RS construction");<a name="line.800"></a> -<span class="sourceLineNo">801</span> }<a name="line.801"></a> -<span class="sourceLineNo">802</span> }<a name="line.802"></a> -<span class="sourceLineNo">803</span> }<a name="line.803"></a> -<span class="sourceLineNo">804</span><a name="line.804"></a> -<span class="sourceLineNo">805</span> public String getClusterId() {<a name="line.805"></a> -<span class="sourceLineNo">806</span> return this.clusterId;<a name="line.806"></a> +<span class="sourceLineNo">745</span> protected Class<? extends HttpServlet> getDumpServlet() {<a name="line.745"></a> +<span class="sourceLineNo">746</span> return RSDumpServlet.class;<a name="line.746"></a> +<span class="sourceLineNo">747</span> }<a name="line.747"></a> +<span class="sourceLineNo">748</span><a name="line.748"></a> +<span class="sourceLineNo">749</span> @Override<a name="line.749"></a> +<span class="sourceLineNo">750</span> public boolean registerService(com.google.protobuf.Service instance) {<a name="line.750"></a> +<span class="sourceLineNo">751</span> /*<a name="line.751"></a> +<span class="sourceLineNo">752</span> * No stacking of instances is allowed for a single executorService name<a name="line.752"></a> +<span class="sourceLineNo">753</span> */<a name="line.753"></a> +<span class="sourceLineNo">754</span> com.google.protobuf.Descriptors.ServiceDescriptor serviceDesc =<a name="line.754"></a> +<span class="sourceLineNo">755</span> instance.getDescriptorForType();<a name="line.755"></a> +<span class="sourceLineNo">756</span> String serviceName = CoprocessorRpcUtils.getServiceName(serviceDesc);<a name="line.756"></a> +<span class="sourceLineNo">757</span> if (coprocessorServiceHandlers.containsKey(serviceName)) {<a name="line.757"></a> +<span class="sourceLineNo">758</span> LOG.error("Coprocessor executorService " + serviceName<a name="line.758"></a> +<span class="sourceLineNo">759</span> + " already registered, rejecting request from " + instance);<a name="line.759"></a> +<span class="sourceLineNo">760</span> return false;<a name="line.760"></a> +<span class="sourceLineNo">761</span> }<a name="line.761"></a> +<span class="sourceLineNo">762</span><a name="line.762"></a> +<span class="sourceLineNo">763</span> coprocessorServiceHandlers.put(serviceName, instance);<a name="line.763"></a> +<span class="sourceLineNo">764</span> if (LOG.isDebugEnabled()) {<a name="line.764"></a> +<span class="sourceLineNo">765</span> LOG.debug("Registered regionserver coprocessor executorService: executorService=" + serviceName);<a name="line.765"></a> +<span class="sourceLineNo">766</span> }<a name="line.766"></a> +<span class="sourceLineNo">767</span> return true;<a name="line.767"></a> +<span class="sourceLineNo">768</span> }<a name="line.768"></a> +<span class="sourceLineNo">769</span><a name="line.769"></a> +<span class="sourceLineNo">770</span> /**<a name="line.770"></a> +<span class="sourceLineNo">771</span> * Create a 'smarter' Connection, one that is capable of by-passing RPC if the request is to<a name="line.771"></a> +<span class="sourceLineNo">772</span> * the local server; i.e. a short-circuit Connection. Safe to use going to local or remote<a name="line.772"></a> +<span class="sourceLineNo">773</span> * server. Create this instance in a method can be intercepted and mocked in tests.<a name="line.773"></a> +<span class="sourceLineNo">774</span> * @throws IOException<a name="line.774"></a> +<span class="sourceLineNo">775</span> */<a name="line.775"></a> +<span class="sourceLineNo">776</span> @VisibleForTesting<a name="line.776"></a> +<span class="sourceLineNo">777</span> protected ClusterConnection createClusterConnection() throws IOException {<a name="line.777"></a> +<span class="sourceLineNo">778</span> Configuration conf = this.conf;<a name="line.778"></a> +<span class="sourceLineNo">779</span> if (conf.get(HConstants.CLIENT_ZOOKEEPER_QUORUM) != null) {<a name="line.779"></a> +<span class="sourceLineNo">780</span> // Use server ZK cluster for server-issued connections, so we clone<a name="line.780"></a> +<span class="sourceLineNo">781</span> // the conf and unset the client ZK related properties<a name="line.781"></a> +<span class="sourceLineNo">782</span> conf = new Configuration(this.conf);<a name="line.782"></a> +<span class="sourceLineNo">783</span> conf.unset(HConstants.CLIENT_ZOOKEEPER_QUORUM);<a name="line.783"></a> +<span class="sourceLineNo">784</span> }<a name="line.784"></a> +<span class="sourceLineNo">785</span> // Create a cluster connection that when appropriate, can short-circuit and go directly to the<a name="line.785"></a> +<span class="sourceLineNo">786</span> // local server if the request is to the local server bypassing RPC. Can be used for both local<a name="line.786"></a> +<span class="sourceLineNo">787</span> // and remote invocations.<a name="line.787"></a> +<span class="sourceLineNo">788</span> return ConnectionUtils.createShortCircuitConnection(conf, null, userProvider.getCurrent(),<a name="line.788"></a> +<span class="sourceLineNo">789</span> serverName, rpcServices, rpcServices);<a name="line.789"></a> +<span class="sourceLineNo">790</span> }<a name="line.790"></a> +<span class="sourceLineNo">791</span><a name="line.791"></a> +<span class="sourceLineNo">792</span> /**<a name="line.792"></a> +<span class="sourceLineNo">793</span> * Run test on configured codecs to make sure supporting libs are in place.<a name="line.793"></a> +<span class="sourceLineNo">794</span> * @param c<a name="line.794"></a> +<span class="sourceLineNo">795</span> * @throws IOException<a name="line.795"></a> +<span class="sourceLineNo">796</span> */<a name="line.796"></a> +<span class="sourceLineNo">797</span> private static void checkCodecs(final Configuration c) throws IOException {<a name="line.797"></a> +<span class="sourceLineNo">798</span> // check to see if the codec list is available:<a name="line.798"></a> +<span class="sourceLineNo">799</span> String [] codecs = c.getStrings("hbase.regionserver.codecs", (String[])null);<a name="line.799"></a> +<span class="sourceLineNo">800</span> if (codecs == null) return;<a name="line.800"></a> +<span class="sourceLineNo">801</span> for (String codec : codecs) {<a name="line.801"></a> +<span class="sourceLineNo">802</span> if (!CompressionTest.testCompression(codec)) {<a name="line.802"></a> +<span class="sourceLineNo">803</span> throw new IOException("Compression codec " + codec +<a name="line.803"></a> +<span class="sourceLineNo">804</span> " not supported, aborting RS construction");<a name="line.804"></a> +<span class="sourceLineNo">805</span> }<a name="line.805"></a> +<span class="sourceLineNo">806</span> }<a name="line.806"></a> <span class="sourceLineNo">807</span> }<a name="line.807"></a> <span class="sourceLineNo">808</span><a name="line.808"></a> -<span class="sourceLineNo">809</span> /**<a name="line.809"></a> -<span class="sourceLineNo">810</span> * Setup our cluster connection if not already initialized.<a name="line.810"></a> -<span class="sourceLineNo">811</span> * @throws IOException<a name="line.811"></a> -<span class="sourceLineNo">812</span> */<a name="line.812"></a> -<span class="sourceLineNo">813</span> protected synchronized void setupClusterConnection() throws IOException {<a name="line.813"></a> -<span class="sourceLineNo">814</span> if (clusterConnection == null) {<a name="line.814"></a> -<span class="sourceLineNo">815</span> clusterConnection = createClusterConnection();<a name="line.815"></a> -<span class="sourceLineNo">816</span> metaTableLocator = new MetaTableLocator();<a name="line.816"></a> -<span class="sourceLineNo">817</span> }<a name="line.817"></a> -<span class="sourceLineNo">818</span> }<a name="line.818"></a> -<span class="sourceLineNo">819</span><a name="line.819"></a> -<span class="sourceLineNo">820</span> /**<a name="line.820"></a> -<span class="sourceLineNo">821</span> * All initialization needed before we go register with Master.<br><a name="line.821"></a> -<span class="sourceLineNo">822</span> * Do bare minimum. Do bulk of initializations AFTER we've connected to the Master.<br><a name="line.822"></a> -<span class="sourceLineNo">823</span> * In here we just put up the RpcServer, setup Connection, and ZooKeeper.<a name="line.823"></a> -<span class="sourceLineNo">824</span> */<a name="line.824"></a> -<span class="sourceLineNo">825</span> private void preRegistrationInitialization() {<a name="line.825"></a> -<span class="sourceLineNo">826</span> try {<a name="line.826"></a> -<span class="sourceLineNo">827</span> initializeZooKeeper();<a name="line.827"></a> -<span class="sourceLineNo">828</span> setupClusterConnection();<a name="line.828"></a> -<span class="sourceLineNo">829</span> // Setup RPC client for master communication<a name="line.829"></a> -<span class="sourceLineNo">830</span> this.rpcClient = RpcClientFactory.createClient(conf, clusterId, new InetSocketAddress(<a name="line.830"></a> -<span class="sourceLineNo">831</span> this.rpcServices.isa.getAddress(), 0), clusterConnection.getConnectionMetrics());<a name="line.831"></a> -<span class="sourceLineNo">832</span> } catch (Throwable t) {<a name="line.832"></a> -<span class="sourceLineNo">833</span> // Call stop if error or process will stick around for ever since server<a name="line.833"></a> -<span class="sourceLineNo">834</span> // puts up non-daemon threads.<a name="line.834"></a> -<span class="sourceLineNo">835</span> this.rpcServices.stop();<a name="line.835"></a> -<span class="sourceLineNo">836</span> abort("Initialization of RS failed. Hence aborting RS.", t);<a name="line.836"></a> -<span class="sourceLineNo">837</span> }<a name="line.837"></a> -<span class="sourceLineNo">838</span> }<a name="line.838"></a> -<span class="sourceLineNo">839</span><a name="line.839"></a> -<span class="sourceLineNo">840</span> /**<a name="line.840"></a> -<span class="sourceLineNo">841</span> * Bring up connection to zk ensemble and then wait until a master for this cluster and then after<a name="line.841"></a> -<span class="sourceLineNo">842</span> * that, wait until cluster 'up' flag has been set. This is the order in which master does things.<a name="line.842"></a> -<span class="sourceLineNo">843</span> * <p><a name="line.843"></a> -<span class="sourceLineNo">844</span> * Finally open long-living server short-circuit connection.<a name="line.844"></a> -<span class="sourceLineNo">845</span> */<a name="line.845"></a> -<span class="sourceLineNo">846</span> @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE",<a name="line.846"></a> -<span class="sourceLineNo">847</span> justification="cluster Id znode read would give us correct response")<a name="line.847"></a> -<span class="sourceLineNo">848</span> private void initializeZooKeeper() throws IOException, InterruptedException {<a name="line.848"></a> -<span class="sourceLineNo">849</span> // Nothing to do in here if no Master in the mix.<a name="line.849"></a> -<span class="sourceLineNo">850</span> if (this.masterless) {<a name="line.850"></a> -<span class="sourceLineNo">851</span> return;<a name="line.851"></a> -<span class="sourceLineNo">852</span> }<a name="line.852"></a> -<span class="sourceLineNo">853</span><a name="line.853"></a> -<span class="sourceLineNo">854</span> // Create the master address tracker, register with zk, and start it. Then<a name="line.854"></a> -<span class="sourceLineNo">855</span> // block until a master is available. No point in starting up if no master<a name="line.855"></a> -<span class="sourceLineNo">856</span> // running.<a name="line.856"></a> -<span class="sourceLineNo">857</span> blockAndCheckIfStopped(this.masterAddressTracker);<a name="line.857"></a> -<span class="sourceLineNo">858</span><a name="line.858"></a> -<span class="sourceLineNo">859</span> // Wait on cluster being up. Master will set this flag up in zookeeper<a name="line.859"></a> -<span class="sourceLineNo">860</span> // when ready.<a name="line.860"></a> -<span class="sourceLineNo">861</span> blockAndCheckIfStopped(this.clusterStatusTracker);<a name="line.861"></a> +<span class="sourceLineNo">809</span> public String getClusterId() {<a name="line.809"></a> +<span class="sourceLineNo">810</span> return this.clusterId;<a name="line.810"></a> +<span class="sourceLineNo">811</span> }<a name="line.811"></a> +<span class="sourceLineNo">812</span><a name="line.812"></a> +<span class="sourceLineNo">813</span> /**<a name="line.813"></a> +<span class="sourceLineNo">814</span> * Setup our cluster connection if not already initialized.<a name="line.814"></a> +<span class="sourceLineNo">815</span> * @throws IOException<a name="line.815"></a> +<span class="sourceLineNo">816</span> */<a name="line.816"></a> +<span class="sourceLineNo">817</span> protected synchronized void setupClusterConnection() throws IOException {<a name="line.817"></a> +<span class="sourceLineNo">818</span> if (clusterConnection == null) {<a name="line.818"></a> +<span class="sourceLineNo">819</span> clusterConnection = createClusterConnection();<a name="line.819"></a> +<span class="sourceLineNo">820</span> metaTableLocator = new MetaTableLocator();<a name="line.820"></a> +<span class="sourceLineNo">821</span> }<a name="line.821"></a> +<span class="sourceLineNo">822</span> }<a name="line.822"></a> +<span class="sourceLineNo">823</span><a name="line.823"></a> +<span class="sourceLineNo">824</span> /**<a name="line.824"></a> +<span class="sourceLineNo">825</span> * All initialization needed before we go register with Master.<br><a name="line.825"></a> +<span class="sourceLineNo">826</span> * Do bare minimum. Do bulk of initializations AFTER we've connected to the Master.<br><a name="line.826"></a> +<span class="sourceLineNo">827</span> * In here we just put up the RpcServer, setup Connection, and ZooKeeper.<a name="line.827"></a> +<span class="sourceLineNo">828</span> */<a name="line.828"></a> +<span class="sourceLineNo">829</span> private void preRegistrationInitialization() {<a name="line.829"></a> +<span class="sourceLineNo">830</span> try {<a name="line.830"></a> +<span class="sourceLineNo">831</span> initializeZooKeeper();<a name="line.831"></a> +<span class="sourceLineNo">832</span> setupClusterConnection();<a name="line.832"></a> +<span class="sourceLineNo">833</span> // Setup RPC client for master communication<a name="line.833"></a> +<span class="sourceLineNo">834</span> this.rpcClient = RpcClientFactory.createClient(conf, clusterId, new InetSocketAddress(<a name="line.834"></a> +<span class="sourceLineNo">835</span> this.rpcServices.isa.getAddress(), 0), clusterConnection.getConnectionMetrics());<a name="line.835"></a> +<span class="sourceLineNo">836</span> } catch (Throwable t) {<a name="line.836"></a> +<span class="sourceLineNo">837</span> // Call stop if error or process will stick around for ever since server<a name="line.837"></a> +<span class="sourceLineNo">838</span> // puts up non-daemon threads.<a name="line.838"></a> +<span class="sourceLineNo">839</span> this.rpcServices.stop();<a name="line.839"></a> +<span class="sourceLineNo">840</span> abort("Initialization of RS failed. Hence aborting RS.", t);<a name="line.840"></a> +<span class="sourceLineNo">841</span> }<a name="line.841"></a> +<span class="sourceLineNo">842</span> }<a name="line.842"></a> +<span class="sourceLineNo">843</span><a name="line.843"></a> +<span class="sourceLineNo">844</span> /**<a name="line.844"></a> +<span class="sourceLineNo">845</span> * Bring up connection to zk ensemble and then wait until a master for this cluster and then after<a name="line.845"></a> +<span class="sourceLineNo">846</span> * that, wait until cluster 'up' flag has been set. This is the order in which master does things.<a name="line.846"></a> +<span class="sourceLineNo">847</span> * <p><a name="line.847"></a> +<span class="sourceLineNo">848</span> * Finally open long-living server short-circuit connection.<a name="line.848"></a> +<span class="sourceLineNo">849</span> */<a name="line.849"></a> +<span class="sourceLineNo">850</span> @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE",<a name="line.850"></a> +<span class="sourceLineNo">851</span> justification="cluster Id znode read would give us correct response")<a name="line.851"></a> +<span class="sourceLineNo">852</span> private void initializeZooKeeper() throws IOException, InterruptedException {<a name="line.852"></a> +<span class="sourceLineNo">853</span> // Nothing to do in here if no Master in the mix.<a name="line.853"></a> +<span class="sourceLineNo">854</span> if (this.masterless) {<a name="line.854"></a> +<span class="sourceLineNo">855</span> return;<a name="line.855"></a> +<span class="sourceLineNo">856</span> }<a name="line.856"></a> +<span class="sourceLineNo">857</span><a name="line.857"></a> +<span class="sourceLineNo">858</span> // Create the master address tracker, register with zk, and start it. Then<a name="line.858"></a> +<span class="sourceLineNo">859</span> // block until a master is available. No point in starting up if no master<a name="line.859"></a> +<span class="sourceLineNo">860</span> // running.<a name="line.860"></a> +<span class="sourceLineNo">861</span> blockAndCheckIfStopped(this.masterAddressTracker);<a name="line.861"></a> <span class="sourceLineNo">862</span><a name="line.862"></a> -<span class="sourceLineNo">863</span> // If we are HMaster then the cluster id should have already been set.<a name="line.863"></a> -<span class="sourceLineNo">864</span> if (clusterId == null) {<a name="line.864"></a> -<span class="sourceLineNo">865</span> // Retrieve clusterId<a name="line.865"></a> -<span class="sourceLineNo">866</span> // Since cluster status is now up<a name="line.866"></a> -<span class="sourceLineNo">867</span> // ID should have already been set by HMaster<a name="line.867"></a> -<span class="sourceLineNo">868</span> try {<a name="line.868"></a> -<span class="sourceLineNo">869</span> clusterId = ZKClusterId.readClusterIdZNode(this.zooKeeper);<a name="line.869"></a> -<span class="sourceLineNo">870</span> if (clusterId == null) {<a name="line.870"></a> -<span class="sourceLineNo">871</span> this.abort("Cluster ID has not been set");<a name="line.871"></a> -<span class="sourceLineNo">872</span> }<a name="line.872"></a> -<span class="sourceLineNo">873</span> LOG.info("ClusterId : " + clusterId);<a name="line.873"></a> -<span class="sourceLineNo">874</span> } catch (KeeperException e) {<a name="line.874"></a> -<span class="sourceLineNo">875</span> this.abort("Failed to retrieve Cluster ID", e);<a name="line.875"></a> -<span class="sourceLineNo">876</span> }<a name="line.876"></a> -<span class="sourceLineNo">877</span> }<a name="line.877"></a> -<span class="sourceLineNo">878</span><a name="line.878"></a> -<span class="sourceLineNo">879</span> // In case colocated master, wait here till it's active.<a name="line.879"></a> -<span class="sourceLineNo">880</span> // So backup masters won't start as regionservers.<a name="line.880"></a> -<span class="sourceLineNo">881</span> // This is to avoid showing backup masters as regionservers<a name="line.881"></a> -<span class="sourceLineNo">882</span> // in master web UI, or assigning any region to them.<a name="line.882"></a> +<span class="sourceLineNo">863</span> // Wait on cluster being up. Master will set this flag up in zookeeper<a name="line.863"></a> +<span class="sourceLineNo">864</span> // when ready.<a name="line.864"></a> +<span class="sourceLineNo">865</span> blockAndCheckIfStopped(this.clusterStatusTracker);<a name="line.865"></a> +<span class="sourceLineNo">866</span><a name="line.866"></a> +<span class="sourceLineNo">867</span> // If we are HMaster then the cluster id should have already been set.<a name="line.867"></a> +<span class="sourceLineNo">868</span> if (clusterId == null) {<a name="line.868"></a> +<span class="sourceLineNo">869</span> // Retrieve clusterId<a name="line.869"></a> +<span class="sourceLineNo">870</span> // Since cluster status is now up<a name="line.870"></a> +<span class="sourceLineNo">871</span> // ID should have already been set by HMaster<a name="line.871"></a> +<span class="sourceLineNo">872</span> try {<a name="line.872"></a> +<span class="sourceLineNo">873</span> clusterId = ZKClusterId.readClusterIdZNode(this.zooKeeper);<a name="line.873"></a> +<span class="sourceLineNo">874</span> if (clusterId == null) {<a name="line.874"></a> +<span class="sourceLineNo">875</span> this.abort("Cluster ID has not been set");<a name="line.875"></a> +<span class="sourceLineNo">876</span> }<a name="line.876"></a> +<span class="sourceLineNo">877</span> LOG.info("ClusterId : " + clusterId);<a name="line.877"></a> +<span class="sourceLineNo">878</span> } catch (KeeperException e) {<a name="line.878"></a> +<span class="sourceLineNo">879</span> this.abort("Failed to retrieve Cluster ID", e);<a name="line.879"></a> +<span class="sourceLineNo">880</span> }<a name="line.880"></a> +<span class="sourceLineNo">881</span> }<a name="line.881"></a> +<span class="sourceLineNo">882</span><a name="line.882"></a> <span class="sourceLineNo">883</span> waitForMasterActive();<a name="line.883"></a> <span class="sourceLineNo">884</span> if (isStopped() || isAborted()) {<a name="line.884"></a> <span class="sourceLineNo">885</span> return; // No need for further initialization<a name="line.885"></a>
http://git-wip-us.apache.org/repos/asf/hbase-site/blob/738e976e/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HStore.StoreFlusherImpl.html ---------------------------------------------------------------------- diff --git a/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HStore.StoreFlusherImpl.html b/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HStore.StoreFlusherImpl.html index 34b8b52..3f9d8d2 100644 --- a/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HStore.StoreFlusherImpl.html +++ b/devapidocs/src-html/org/apache/hadoop/hbase/regionserver/HStore.StoreFlusherImpl.html @@ -935,7 +935,7 @@ <span class="sourceLineNo">927</span> storeEngine.getStoreFileManager().clearCompactedFiles();<a name="line.927"></a> <span class="sourceLineNo">928</span> // clear the compacted files<a name="line.928"></a> <span class="sourceLineNo">929</span> if (CollectionUtils.isNotEmpty(compactedfiles)) {<a name="line.929"></a> -<span class="sourceLineNo">930</span> removeCompactedfiles(compactedfiles);<a name="line.930"></a> +<span class="sourceLineNo">930</span> removeCompactedfiles(compactedfiles, true);<a name="line.930"></a> <span class="sourceLineNo">931</span> }<a name="line.931"></a> <span class="sourceLineNo">932</span> if (!result.isEmpty()) {<a name="line.932"></a> <span class="sourceLineNo">933</span> // initialize the thread pool for closing store files in parallel.<a name="line.933"></a> @@ -2583,190 +2583,221 @@ <span class="sourceLineNo">2575</span> * Closes and archives the compacted files under this store<a name="line.2575"></a> <span class="sourceLineNo">2576</span> */<a name="line.2576"></a> <span class="sourceLineNo">2577</span> public synchronized void closeAndArchiveCompactedFiles() throws IOException {<a name="line.2577"></a> -<span class="sourceLineNo">2578</span> // ensure other threads do not attempt to archive the same files on close()<a name="line.2578"></a> -<span class="sourceLineNo">2579</span> archiveLock.lock();<a name="line.2579"></a> -<span class="sourceLineNo">2580</span> try {<a name="line.2580"></a> -<span class="sourceLineNo">2581</span> lock.readLock().lock();<a name="line.2581"></a> -<span class="sourceLineNo">2582</span> Collection<HStoreFile> copyCompactedfiles = null;<a name="line.2582"></a> -<span class="sourceLineNo">2583</span> try {<a name="line.2583"></a> -<span class="sourceLineNo">2584</span> Collection<HStoreFile> compactedfiles =<a name="line.2584"></a> -<span class="sourceLineNo">2585</span> this.getStoreEngine().getStoreFileManager().getCompactedfiles();<a name="line.2585"></a> -<span class="sourceLineNo">2586</span> if (CollectionUtils.isNotEmpty(compactedfiles)) {<a name="line.2586"></a> -<span class="sourceLineNo">2587</span> // Do a copy under read lock<a name="line.2587"></a> -<span class="sourceLineNo">2588</span> copyCompactedfiles = new ArrayList<>(compactedfiles);<a name="line.2588"></a> -<span class="sourceLineNo">2589</span> } else {<a name="line.2589"></a> -<span class="sourceLineNo">2590</span> LOG.trace("No compacted files to archive");<a name="line.2590"></a> -<span class="sourceLineNo">2591</span> }<a name="line.2591"></a> -<span class="sourceLineNo">2592</span> } finally {<a name="line.2592"></a> -<span class="sourceLineNo">2593</span> lock.readLock().unlock();<a name="line.2593"></a> -<span class="sourceLineNo">2594</span> }<a name="line.2594"></a> -<span class="sourceLineNo">2595</span> if (CollectionUtils.isNotEmpty(copyCompactedfiles)) {<a name="line.2595"></a> -<span class="sourceLineNo">2596</span> removeCompactedfiles(copyCompactedfiles);<a name="line.2596"></a> -<span class="sourceLineNo">2597</span> }<a name="line.2597"></a> -<span class="sourceLineNo">2598</span> } finally {<a name="line.2598"></a> -<span class="sourceLineNo">2599</span> archiveLock.unlock();<a name="line.2599"></a> -<span class="sourceLineNo">2600</span> }<a name="line.2600"></a> -<span class="sourceLineNo">2601</span> }<a name="line.2601"></a> -<span class="sourceLineNo">2602</span><a name="line.2602"></a> -<span class="sourceLineNo">2603</span> /**<a name="line.2603"></a> -<span class="sourceLineNo">2604</span> * Archives and removes the compacted files<a name="line.2604"></a> -<span class="sourceLineNo">2605</span> * @param compactedfiles The compacted files in this store that are not active in reads<a name="line.2605"></a> -<span class="sourceLineNo">2606</span> */<a name="line.2606"></a> -<span class="sourceLineNo">2607</span> private void removeCompactedfiles(Collection<HStoreFile> compactedfiles)<a name="line.2607"></a> -<span class="sourceLineNo">2608</span> throws IOException {<a name="line.2608"></a> -<span class="sourceLineNo">2609</span> final List<HStoreFile> filesToRemove = new ArrayList<>(compactedfiles.size());<a name="line.2609"></a> -<span class="sourceLineNo">2610</span> final List<Long> storeFileSizes = new ArrayList<>(compactedfiles.size());<a name="line.2610"></a> -<span class="sourceLineNo">2611</span> for (final HStoreFile file : compactedfiles) {<a name="line.2611"></a> -<span class="sourceLineNo">2612</span> synchronized (file) {<a name="line.2612"></a> -<span class="sourceLineNo">2613</span> try {<a name="line.2613"></a> -<span class="sourceLineNo">2614</span> StoreFileReader r = file.getReader();<a name="line.2614"></a> -<span class="sourceLineNo">2615</span> if (r == null) {<a name="line.2615"></a> -<span class="sourceLineNo">2616</span> LOG.debug("The file {} was closed but still not archived", file);<a name="line.2616"></a> -<span class="sourceLineNo">2617</span> // HACK: Temporarily re-open the reader so we can get the size of the file. Ideally,<a name="line.2617"></a> -<span class="sourceLineNo">2618</span> // we should know the size of an HStoreFile without having to ask the HStoreFileReader<a name="line.2618"></a> -<span class="sourceLineNo">2619</span> // for that.<a name="line.2619"></a> -<span class="sourceLineNo">2620</span> long length = getStoreFileSize(file);<a name="line.2620"></a> -<span class="sourceLineNo">2621</span> filesToRemove.add(file);<a name="line.2621"></a> -<span class="sourceLineNo">2622</span> storeFileSizes.add(length);<a name="line.2622"></a> -<span class="sourceLineNo">2623</span> continue;<a name="line.2623"></a> -<span class="sourceLineNo">2624</span> }<a name="line.2624"></a> -<span class="sourceLineNo">2625</span> if (file.isCompactedAway() && !file.isReferencedInReads()) {<a name="line.2625"></a> -<span class="sourceLineNo">2626</span> // Even if deleting fails we need not bother as any new scanners won't be<a name="line.2626"></a> -<span class="sourceLineNo">2627</span> // able to use the compacted file as the status is already compactedAway<a name="line.2627"></a> -<span class="sourceLineNo">2628</span> LOG.trace("Closing and archiving the file {}", file);<a name="line.2628"></a> -<span class="sourceLineNo">2629</span> // Copy the file size before closing the reader<a name="line.2629"></a> -<span class="sourceLineNo">2630</span> final long length = r.length();<a name="line.2630"></a> -<span class="sourceLineNo">2631</span> r.close(true);<a name="line.2631"></a> -<span class="sourceLineNo">2632</span> // Just close and return<a name="line.2632"></a> -<span class="sourceLineNo">2633</span> filesToRemove.add(file);<a name="line.2633"></a> -<span class="sourceLineNo">2634</span> // Only add the length if we successfully added the file to `filesToRemove`<a name="line.2634"></a> -<span class="sourceLineNo">2635</span> storeFileSizes.add(length);<a name="line.2635"></a> -<span class="sourceLineNo">2636</span> } else {<a name="line.2636"></a> -<span class="sourceLineNo">2637</span> LOG.info("Can't archive compacted file " + file.getPath()<a name="line.2637"></a> -<span class="sourceLineNo">2638</span> + " because of either isCompactedAway = " + file.isCompactedAway()<a name="line.2638"></a> -<span class="sourceLineNo">2639</span> + " or file has reference, isReferencedInReads = " + file.isReferencedInReads()<a name="line.2639"></a> -<span class="sourceLineNo">2640</span> + ", skipping for now.");<a name="line.2640"></a> -<span class="sourceLineNo">2641</span> }<a name="line.2641"></a> -<span class="sourceLineNo">2642</span> } catch (Exception e) {<a name="line.2642"></a> -<span class="sourceLineNo">2643</span> LOG.error("Exception while trying to close the compacted store file {}",<a name="line.2643"></a> -<span class="sourceLineNo">2644</span> file.getPath(), e);<a name="line.2644"></a> -<span class="sourceLineNo">2645</span> }<a name="line.2645"></a> -<span class="sourceLineNo">2646</span> }<a name="line.2646"></a> -<span class="sourceLineNo">2647</span> }<a name="line.2647"></a> -<span class="sourceLineNo">2648</span> if (this.isPrimaryReplicaStore()) {<a name="line.2648"></a> -<span class="sourceLineNo">2649</span> // Only the primary region is allowed to move the file to archive.<a name="line.2649"></a> -<span class="sourceLineNo">2650</span> // The secondary region does not move the files to archive. Any active reads from<a name="line.2650"></a> -<span class="sourceLineNo">2651</span> // the secondary region will still work because the file as such has active readers on it.<a name="line.2651"></a> -<span class="sourceLineNo">2652</span> if (!filesToRemove.isEmpty()) {<a name="line.2652"></a> -<span class="sourceLineNo">2653</span> LOG.debug("Moving the files {} to archive", filesToRemove);<a name="line.2653"></a> -<span class="sourceLineNo">2654</span> // Only if this is successful it has to be removed<a name="line.2654"></a> -<span class="sourceLineNo">2655</span> try {<a name="line.2655"></a> -<span class="sourceLineNo">2656</span> this.fs.removeStoreFiles(this.getColumnFamilyDescriptor().getNameAsString(), filesToRemove);<a name="line.2656"></a> -<span class="sourceLineNo">2657</span> } catch (FailedArchiveException fae) {<a name="line.2657"></a> -<span class="sourceLineNo">2658</span> // Even if archiving some files failed, we still need to clear out any of the<a name="line.2658"></a> -<span class="sourceLineNo">2659</span> // files which were successfully archived. Otherwise we will receive a<a name="line.2659"></a> -<span class="sourceLineNo">2660</span> // FileNotFoundException when we attempt to re-archive them in the next go around.<a name="line.2660"></a> -<span class="sourceLineNo">2661</span> Collection<Path> failedFiles = fae.getFailedFiles();<a name="line.2661"></a> -<span class="sourceLineNo">2662</span> Iterator<HStoreFile> iter = filesToRemove.iterator();<a name="line.2662"></a> -<span class="sourceLineNo">2663</span> Iterator<Long> sizeIter = storeFileSizes.iterator();<a name="line.2663"></a> -<span class="sourceLineNo">2664</span> while (iter.hasNext()) {<a name="line.2664"></a> -<span class="sourceLineNo">2665</span> sizeIter.next();<a name="line.2665"></a> -<span class="sourceLineNo">2666</span> if (failedFiles.contains(iter.next().getPath())) {<a name="line.2666"></a> -<span class="sourceLineNo">2667</span> iter.remove();<a name="line.2667"></a> -<span class="sourceLineNo">2668</span> sizeIter.remove();<a name="line.2668"></a> -<span class="sourceLineNo">2669</span> }<a name="line.2669"></a> +<span class="sourceLineNo">2578</span> closeAndArchiveCompactedFiles(false);<a name="line.2578"></a> +<span class="sourceLineNo">2579</span> }<a name="line.2579"></a> +<span class="sourceLineNo">2580</span><a name="line.2580"></a> +<span class="sourceLineNo">2581</span> @VisibleForTesting<a name="line.2581"></a> +<span class="sourceLineNo">2582</span> public synchronized void closeAndArchiveCompactedFiles(boolean storeClosing) throws IOException {<a name="line.2582"></a> +<span class="sourceLineNo">2583</span> // ensure other threads do not attempt to archive the same files on close()<a name="line.2583"></a> +<span class="sourceLineNo">2584</span> archiveLock.lock();<a name="line.2584"></a> +<span class="sourceLineNo">2585</span> try {<a name="line.2585"></a> +<span class="sourceLineNo">2586</span> lock.readLock().lock();<a name="line.2586"></a> +<span class="sourceLineNo">2587</span> Collection<HStoreFile> copyCompactedfiles = null;<a name="line.2587"></a> +<span class="sourceLineNo">2588</span> try {<a name="line.2588"></a> +<span class="sourceLineNo">2589</span> Collection<HStoreFile> compactedfiles =<a name="line.2589"></a> +<span class="sourceLineNo">2590</span> this.getStoreEngine().getStoreFileManager().getCompactedfiles();<a name="line.2590"></a> +<span class="sourceLineNo">2591</span> if (CollectionUtils.isNotEmpty(compactedfiles)) {<a name="line.2591"></a> +<span class="sourceLineNo">2592</span> // Do a copy under read lock<a name="line.2592"></a> +<span class="sourceLineNo">2593</span> copyCompactedfiles = new ArrayList<>(compactedfiles);<a name="line.2593"></a> +<span class="sourceLineNo">2594</span> } else {<a name="line.2594"></a> +<span class="sourceLineNo">2595</span> LOG.trace("No compacted files to archive");<a name="line.2595"></a> +<span class="sourceLineNo">2596</span> }<a name="line.2596"></a> +<span class="sourceLineNo">2597</span> } finally {<a name="line.2597"></a> +<span class="sourceLineNo">2598</span> lock.readLock().unlock();<a name="line.2598"></a> +<span class="sourceLineNo">2599</span> }<a name="line.2599"></a> +<span class="sourceLineNo">2600</span> if (CollectionUtils.isNotEmpty(copyCompactedfiles)) {<a name="line.2600"></a> +<span class="sourceLineNo">2601</span> removeCompactedfiles(copyCompactedfiles, storeClosing);<a name="line.2601"></a> +<span class="sourceLineNo">2602</span> }<a name="line.2602"></a> +<span class="sourceLineNo">2603</span> } finally {<a name="line.2603"></a> +<span class="sourceLineNo">2604</span> archiveLock.unlock();<a name="line.2604"></a> +<span class="sourceLineNo">2605</span> }<a name="line.2605"></a> +<span class="sourceLineNo">2606</span> }<a name="line.2606"></a> +<span class="sourceLineNo">2607</span><a name="line.2607"></a> +<span class="sourceLineNo">2608</span> /**<a name="line.2608"></a> +<span class="sourceLineNo">2609</span> * Archives and removes the compacted files<a name="line.2609"></a> +<span class="sourceLineNo">2610</span> * @param compactedfiles The compacted files in this store that are not active in reads<a name="line.2610"></a> +<span class="sourceLineNo">2611</span> */<a name="line.2611"></a> +<span class="sourceLineNo">2612</span> private void removeCompactedfiles(Collection<HStoreFile> compactedfiles, boolean storeClosing)<a name="line.2612"></a> +<span class="sourceLineNo">2613</span> throws IOException {<a name="line.2613"></a> +<span class="sourceLineNo">2614</span> final List<HStoreFile> filesToRemove = new ArrayList<>(compactedfiles.size());<a name="line.2614"></a> +<span class="sourceLineNo">2615</span> final List<Long> storeFileSizes = new ArrayList<>(compactedfiles.size());<a name="line.2615"></a> +<span class="sourceLineNo">2616</span> for (final HStoreFile file : compactedfiles) {<a name="line.2616"></a> +<span class="sourceLineNo">2617</span> synchronized (file) {<a name="line.2617"></a> +<span class="sourceLineNo">2618</span> try {<a name="line.2618"></a> +<span class="sourceLineNo">2619</span> StoreFileReader r = file.getReader();<a name="line.2619"></a> +<span class="sourceLineNo">2620</span> if (r == null) {<a name="line.2620"></a> +<span class="sourceLineNo">2621</span> LOG.debug("The file {} was closed but still not archived", file);<a name="line.2621"></a> +<span class="sourceLineNo">2622</span> // HACK: Temporarily re-open the reader so we can get the size of the file. Ideally,<a name="line.2622"></a> +<span class="sourceLineNo">2623</span> // we should know the size of an HStoreFile without having to ask the HStoreFileReader<a name="line.2623"></a> +<span class="sourceLineNo">2624</span> // for that.<a name="line.2624"></a> +<span class="sourceLineNo">2625</span> long length = getStoreFileSize(file);<a name="line.2625"></a> +<span class="sourceLineNo">2626</span> filesToRemove.add(file);<a name="line.2626"></a> +<span class="sourceLineNo">2627</span> storeFileSizes.add(length);<a name="line.2627"></a> +<span class="sourceLineNo">2628</span> continue;<a name="line.2628"></a> +<span class="sourceLineNo">2629</span> }<a name="line.2629"></a> +<span class="sourceLineNo">2630</span><a name="line.2630"></a> +<span class="sourceLineNo">2631</span> //Compacted files in the list should always be marked compacted away. In the event<a name="line.2631"></a> +<span class="sourceLineNo">2632</span> //they're contradicting in order to guarantee data consistency<a name="line.2632"></a> +<span class="sourceLineNo">2633</span> //should we choose one and ignore the other?<a name="line.2633"></a> +<span class="sourceLineNo">2634</span> if (storeClosing && !file.isCompactedAway()) {<a name="line.2634"></a> +<span class="sourceLineNo">2635</span> String msg =<a name="line.2635"></a> +<span class="sourceLineNo">2636</span> "Region closing but StoreFile is in compacted list but not compacted away: " +<a name="line.2636"></a> +<span class="sourceLineNo">2637</span> file.getPath().getName();<a name="line.2637"></a> +<span class="sourceLineNo">2638</span> throw new IllegalStateException(msg);<a name="line.2638"></a> +<span class="sourceLineNo">2639</span> }<a name="line.2639"></a> +<span class="sourceLineNo">2640</span><a name="line.2640"></a> +<span class="sourceLineNo">2641</span> //If store is closing we're ignoring any references to keep things consistent<a name="line.2641"></a> +<span class="sourceLineNo">2642</span> //and remove compacted storefiles from the region directory<a name="line.2642"></a> +<span class="sourceLineNo">2643</span> if (file.isCompactedAway() && (!file.isReferencedInReads() || storeClosing)) {<a name="line.2643"></a> +<span class="sourceLineNo">2644</span> if (storeClosing && file.isReferencedInReads()) {<a name="line.2644"></a> +<span class="sourceLineNo">2645</span> LOG.debug("Region closing but StoreFile still has references: {}",<a name="line.2645"></a> +<span class="sourceLineNo">2646</span> file.getPath().getName());<a name="line.2646"></a> +<span class="sourceLineNo">2647</span> }<a name="line.2647"></a> +<span class="sourceLineNo">2648</span> // Even if deleting fails we need not bother as any new scanners won't be<a name="line.2648"></a> +<span class="sourceLineNo">2649</span> // able to use the compacted file as the status is already compactedAway<a name="line.2649"></a> +<span class="sourceLineNo">2650</span> LOG.trace("Closing and archiving the file {}", file);<a name="line.2650"></a> +<span class="sourceLineNo">2651</span> // Copy the file size before closing the reader<a name="line.2651"></a> +<span class="sourceLineNo">2652</span> final long length = r.length();<a name="line.2652"></a> +<span class="sourceLineNo">2653</span> r.close(true);<a name="line.2653"></a> +<span class="sourceLineNo">2654</span> file.closeStreamReaders(true);<a name="line.2654"></a> +<span class="sourceLineNo">2655</span> // Just close and return<a name="line.2655"></a> +<span class="sourceLineNo">2656</span> filesToRemove.add(file);<a name="line.2656"></a> +<span class="sourceLineNo">2657</span> // Only add the length if we successfully added the file to `filesToRemove`<a name="line.2657"></a> +<span class="sourceLineNo">2658</span> storeFileSizes.add(length);<a name="line.2658"></a> +<span class="sourceLineNo">2659</span> } else {<a name="line.2659"></a> +<span class="sourceLineNo">2660</span> LOG.info("Can't archive compacted file " + file.getPath()<a name="line.2660"></a> +<span class="sourceLineNo">2661</span> + " because of either isCompactedAway = " + file.isCompactedAway()<a name="line.2661"></a> +<span class="sourceLineNo">2662</span> + " or file has reference, isReferencedInReads = " + file.isReferencedInReads()<a name="line.2662"></a> +<span class="sourceLineNo">2663</span> + ", skipping for now.");<a name="line.2663"></a> +<span class="sourceLineNo">2664</span> }<a name="line.2664"></a> +<span class="sourceLineNo">2665</span> } catch (Exception e) {<a name="line.2665"></a> +<span class="sourceLineNo">2666</span> String msg = "Exception while trying to close the compacted store file " +<a name="line.2666"></a> +<span class="sourceLineNo">2667</span> file.getPath().getName();<a name="line.2667"></a> +<span class="sourceLineNo">2668</span> if (storeClosing) {<a name="line.2668"></a> +<span class="sourceLineNo">2669</span> msg = "Store is closing. " + msg;<a name="line.2669"></a> <span class="sourceLineNo">2670</span> }<a name="line.2670"></a> -<span class="sourceLineNo">2671</span> if (!filesToRemove.isEmpty()) {<a name="line.2671"></a> -<span class="sourceLineNo">2672</span> clearCompactedfiles(filesToRemove);<a name="line.2672"></a> -<span class="sourceLineNo">2673</span> }<a name="line.2673"></a> -<span class="sourceLineNo">2674</span> throw fae;<a name="line.2674"></a> -<span class="sourceLineNo">2675</span> }<a name="line.2675"></a> -<span class="sourceLineNo">2676</span> }<a name="line.2676"></a> -<span class="sourceLineNo">2677</span> }<a name="line.2677"></a> -<span class="sourceLineNo">2678</span> if (!filesToRemove.isEmpty()) {<a name="line.2678"></a> -<span class="sourceLineNo">2679</span> // Clear the compactedfiles from the store file manager<a name="line.2679"></a> -<span class="sourceLineNo">2680</span> clearCompactedfiles(filesToRemove);<a name="line.2680"></a> -<span class="sourceLineNo">2681</span> // Try to send report of this archival to the Master for updating quota usage faster<a name="line.2681"></a> -<span class="sourceLineNo">2682</span> reportArchivedFilesForQuota(filesToRemove, storeFileSizes);<a name="line.2682"></a> -<span class="sourceLineNo">2683</span> }<a name="line.2683"></a> -<span class="sourceLineNo">2684</span> }<a name="line.2684"></a> -<span class="sourceLineNo">2685</span><a name="line.2685"></a> -<span class="sourceLineNo">2686</span> /**<a name="line.2686"></a> -<span class="sourceLineNo">2687</span> * Computes the length of a store file without succumbing to any errors along the way. If an<a name="line.2687"></a> -<span class="sourceLineNo">2688</span> * error is encountered, the implementation returns {@code 0} instead of the actual size.<a name="line.2688"></a> -<span class="sourceLineNo">2689</span> *<a name="line.2689"></a> -<span class="sourceLineNo">2690</span> * @param file The file to compute the size of.<a name="line.2690"></a> -<span class="sourceLineNo">2691</span> * @return The size in bytes of the provided {@code file}.<a name="line.2691"></a> -<span class="sourceLineNo">2692</span> */<a name="line.2692"></a> -<span class="sourceLineNo">2693</span> long getStoreFileSize(HStoreFile file) {<a name="line.2693"></a> -<span class="sourceLineNo">2694</span> long length = 0;<a name="line.2694"></a> -<span class="sourceLineNo">2695</span> try {<a name="line.2695"></a> -<span class="sourceLineNo">2696</span> file.initReader();<a name="line.2696"></a> -<span class="sourceLineNo">2697</span> length = file.getReader().length();<a name="line.2697"></a> -<span class="sourceLineNo">2698</span> } catch (IOException e) {<a name="line.2698"></a> -<span class="sourceLineNo">2699</span> LOG.trace("Failed to open reader when trying to compute store file size, ignoring", e);<a name="line.2699"></a> -<span class="sourceLineNo">2700</span> } finally {<a name="line.2700"></a> -<span class="sourceLineNo">2701</span> try {<a name="line.2701"></a> -<span class="sourceLineNo">2702</span> file.closeStoreFile(<a name="line.2702"></a> -<span class="sourceLineNo">2703</span> file.getCacheConf() != null ? file.getCacheConf().shouldEvictOnClose() : true);<a name="line.2703"></a> -<span class="sourceLineNo">2704</span> } catch (IOException e) {<a name="line.2704"></a> -<span class="sourceLineNo">2705</span> LOG.trace("Failed to close reader after computing store file size, ignoring", e);<a name="line.2705"></a> -<span class="sourceLineNo">2706</span> }<a name="line.2706"></a> -<span class="sourceLineNo">2707</span> }<a name="line.2707"></a> -<span class="sourceLineNo">2708</span> return length;<a name="line.2708"></a> -<span class="sourceLineNo">2709</span> }<a name="line.2709"></a> -<span class="sourceLineNo">2710</span><a name="line.2710"></a> -<span class="sourceLineNo">2711</span> public Long preFlushSeqIDEstimation() {<a name="line.2711"></a> -<span class="sourceLineNo">2712</span> return memstore.preFlushSeqIDEstimation();<a name="line.2712"></a> -<span class="sourceLineNo">2713</span> }<a name="line.2713"></a> -<span class="sourceLineNo">2714</span><a name="line.2714"></a> -<span class="sourceLineNo">2715</span> @Override<a name="line.2715"></a> -<span class="sourceLineNo">2716</span> public boolean isSloppyMemStore() {<a name="line.2716"></a> -<span class="sourceLineNo">2717</span> return this.memstore.isSloppy();<a name="line.2717"></a> -<span class="sourceLineNo">2718</span> }<a name="line.2718"></a> -<span class="sourceLineNo">2719</span><a name="line.2719"></a> -<span class="sourceLineNo">2720</span> private void clearCompactedfiles(List<HStoreFile> filesToRemove) throws IOException {<a name="line.2720"></a> -<span class="sourceLineNo">2721</span> LOG.trace("Clearing the compacted file {} from this store", filesToRemove);<a name="line.2721"></a> -<span class="sourceLineNo">2722</span> try {<a name="line.2722"></a> -<span class="sourceLineNo">2723</span> lock.writeLock().lock();<a name="line.2723"></a> -<span class="sourceLineNo">2724</span> this.getStoreEngine().getStoreFileManager().removeCompactedFiles(filesToRemove);<a name="line.2724"></a> -<span class="sourceLineNo">2725</span> } finally {<a name="line.2725"></a> -<span class="sourceLineNo">2726</span> lock.writeLock().unlock();<a name="line.2726"></a> -<span class="sourceLineNo">2727</span> }<a name="line.2727"></a> -<span class="sourceLineNo">2728</span> }<a name="line.2728"></a> -<span class="sourceLineNo">2729</span><a name="line.2729"></a> -<span class="sourceLineNo">2730</span> void reportArchivedFilesForQuota(List<? extends StoreFile> archivedFiles, List<Long> fileSizes) {<a name="line.2730"></a> -<span class="sourceLineNo">2731</span> // Sanity check from the caller<a name="line.2731"></a> -<span class="sourceLineNo">2732</span> if (archivedFiles.size() != fileSizes.size()) {<a name="line.2732"></a> -<span class="sourceLineNo">2733</span> throw new RuntimeException("Coding error: should never see lists of varying size");<a name="line.2733"></a> -<span class="sourceLineNo">2734</span> }<a name="line.2734"></a> -<span class="sourceLineNo">2735</span> RegionServerServices rss = this.region.getRegionServerServices();<a name="line.2735"></a> -<span class="sourceLineNo">2736</span> if (rss == null) {<a name="line.2736"></a> -<span class="sourceLineNo">2737</span> return;<a name="line.2737"></a> +<span class="sourceLineNo">2671</span> LOG.error(msg, e);<a name="line.2671"></a> +<span class="sourceLineNo">2672</span> //if we get an exception let caller know so it can abort the server<a name="line.2672"></a> +<span class="sourceLineNo">2673</span> if (storeClosing) {<a name="line.2673"></a> +<span class="sourceLineNo">2674</span> throw new IOException(msg, e);<a name="line.2674"></a> +<span class="sourceLineNo">2675</span> }<a name="line.2675"></a> +<span class="sourceLineNo">2676</span> }<a name="line.2676"></a> +<span class="sourceLineNo">2677</span> }<a name="line.2677"></a> +<span class="sourceLineNo">2678</span> }<a name="line.2678"></a> +<span class="sourceLineNo">2679</span> if (this.isPrimaryReplicaStore()) {<a name="line.2679"></a> +<span class="sourceLineNo">2680</span> // Only the primary region is allowed to move the file to archive.<a name="line.2680"></a> +<span class="sourceLineNo">2681</span> // The secondary region does not move the files to archive. Any active reads from<a name="line.2681"></a> +<span class="sourceLineNo">2682</span> // the secondary region will still work because the file as such has active readers on it.<a name="line.2682"></a> +<span class="sourceLineNo">2683</span> if (!filesToRemove.isEmpty()) {<a name="line.2683"></a> +<span class="sourceLineNo">2684</span> LOG.debug("Moving the files {} to archive", filesToRemove);<a name="line.2684"></a> +<span class="sourceLineNo">2685</span> // Only if this is successful it has to be removed<a name="line.2685"></a> +<span class="sourceLineNo">2686</span> try {<a name="line.2686"></a> +<span class="sourceLineNo">2687</span> this.fs.removeStoreFiles(this.getColumnFamilyDescriptor().getNameAsString(), filesToRemove);<a name="line.2687"></a> +<span class="sourceLineNo">2688</span> } catch (FailedArchiveException fae) {<a name="line.2688"></a> +<span class="sourceLineNo">2689</span> // Even if archiving some files failed, we still need to clear out any of the<a name="line.2689"></a> +<span class="sourceLineNo">2690</span> // files which were successfully archived. Otherwise we will receive a<a name="line.2690"></a> +<span class="sourceLineNo">2691</span> // FileNotFoundException when we attempt to re-archive them in the next go around.<a name="line.2691"></a> +<span class="sourceLineNo">2692</span> Collection<Path> failedFiles = fae.getFailedFiles();<a name="line.2692"></a> +<span class="sourceLineNo">2693</span> Iterator<HStoreFile> iter = filesToRemove.iterator();<a name="line.2693"></a> +<span class="sourceLineNo">2694</span> Iterator<Long> sizeIter = storeFileSizes.iterator();<a name="line.2694"></a> +<span class="sourceLineNo">2695</span> while (iter.hasNext()) {<a name="line.2695"></a> +<span class="sourceLineNo">2696</span> sizeIter.next();<a name="line.2696"></a> +<span class="sourceLineNo">2697</span> if (failedFiles.contains(iter.next().getPath())) {<a name="line.2697"></a> +<span class="sourceLineNo">2698</span> iter.remove();<a name="line.2698"></a> +<span class="sourceLineNo">2699</span> sizeIter.remove();<a name="line.2699"></a> +<span class="sourceLineNo">2700</span> }<a name="line.2700"></a> +<span class="sourceLineNo">2701</span> }<a name="line.2701"></a> +<span class="sourceLineNo">2702</span> if (!filesToRemove.isEmpty()) {<a name="line.2702"></a> +<span class="sourceLineNo">2703</span> clearCompactedfiles(filesToRemove);<a name="line.2703"></a> +<span class="sourceLineNo">2704</span> }<a name="line.2704"></a> +<span class="sourceLineNo">2705</span> throw fae;<a name="line.2705"></a> +<span class="sourceLineNo">2706</span> }<a name="line.2706"></a> +<span class="sourceLineNo">2707</span> }<a name="line.2707"></a> +<span class="sourceLineNo">2708</span> }<a name="line.2708"></a> +<span class="sourceLineNo">2709</span> if (!filesToRemove.isEmpty()) {<a name="line.2709"></a> +<span class="sourceLineNo">2710</span> // Clear the compactedfiles from the store file manager<a name="line.2710"></a> +<span class="sourceLineNo">2711</span> clearCompactedfiles(filesToRemove);<a name="line.2711"></a> +<span class="sourceLineNo">2712</span> // Try to send report of this archival to the Master for updating quota usage faster<a name="line.2712"></a> +<span class="sourceLineNo">2713</span> reportArchivedFilesForQuota(filesToRemove, storeFileSizes);<a name="line.2713"></a> +<span class="sourceLineNo">2714</span> }<a name="line.2714"></a> +<span class="sourceLineNo">2715</span> }<a name="line.2715"></a> +<span class="sourceLineNo">2716</span><a name="line.2716"></a> +<span class="sourceLineNo">2717</span> /**<a name="line.2717"></a> +<span class="sourceLineNo">2718</span> * Computes the length of a store file without succumbing to any errors along the way. If an<a name="line.2718"></a> +<span class="sourceLineNo">2719</span> * error is encountered, the implementation returns {@code 0} instead of the actual size.<a name="line.2719"></a> +<span class="sourceLineNo">2720</span> *<a name="line.2720"></a> +<span class="sourceLineNo">2721</span> * @param file The file to compute the size of.<a name="line.2721"></a> +<span class="sourceLineNo">2722</span> * @return The size in bytes of the provided {@code file}.<a name="line.2722"></a> +<span class="sourceLineNo">2723</span> */<a name="line.2723"></a> +<span class="sourceLineNo">2724</span> long getStoreFileSize(HStoreFile file) {<a name="line.2724"></a> +<span class="sourceLineNo">2725</span> long length = 0;<a name="line.2725"></a> +<span class="sourceLineNo">2726</span> try {<a name="line.2726"></a> +<span class="sourceLineNo">2727</span> file.initReader();<a name="line.2727"></a> +<span class="sourceLineNo">2728</span> length = file.getReader().length();<a name="line.2728"></a> +<span class="sourceLineNo">2729</span> } catch (IOException e) {<a name="line.2729"></a> +<span class="sourceLineNo">2730</span> LOG.trace("Failed to open reader when trying to compute store file size, ignoring", e);<a name="line.2730"></a> +<span class="sourceLineNo">2731</span> } finally {<a name="line.2731"></a> +<span class="sourceLineNo">2732</span> try {<a name="line.2732"></a> +<span class="sourceLineNo">2733</span> file.closeStoreFile(<a name="line.2733"></a> +<span class="sourceLineNo">2734</span> file.getCacheConf() != null ? file.getCacheConf().shouldEvictOnClose() : true);<a name="line.2734"></a> +<span class="sourceLineNo">2735</span> } catch (IOException e) {<a name="line.2735"></a> +<span class="sourceLineNo">2736</span> LOG.trace("Failed to close reader after computing store file size, ignoring", e);<a name="line.2736"></a> +<span class="sourceLineNo">2737</span> }<a name="line.2737"></a> <span class="sourceLineNo">2738</span> }<a name="line.2738"></a> -<span class="sourceLineNo">2739</span> List<Entry<String,Long>> filesWithSizes = new ArrayList<>(archivedFiles.size());<a name="line.2739"></a> -<span class="sourceLineNo">2740</span> Iterator<Long> fileSizeIter = fileSizes.iterator();<a name="line.2740"></a> -<span class="sourceLineNo">2741</span> for (StoreFile storeFile : archivedFiles) {<a name="line.2741"></a> -<span class="sourceLineNo">2742</span> final long fileSize = fileSizeIter.next();<a name="line.2742"></a> -<span class="sourceLineNo">2743</span> if (storeFile.isHFile() && fileSize != 0) {<a name="line.2743"></a> -<span class="sourceLineNo">2744</span> filesWithSizes.add(Maps.immutableEntry(storeFile.getPath().getName(), fileSize));<a name="line.2744"></a> -<span class="sourceLineNo">2745</span> }<a name="line.2745"></a> -<span class="sourceLineNo">2746</span> }<a name="line.2746"></a> -<span class="sourceLineNo">2747</span> if (LOG.isTraceEnabled()) {<a name="line.2747"></a> -<span class="sourceLineNo">2748</span> LOG.trace("Files archived: " + archivedFiles + ", reporting the following to the Master: "<a name="line.2748"></a> -<span class="sourceLineNo">2749</span> + filesWithSizes);<a name="line.2749"></a> -<span class="sourceLineNo">2750</span> }<a name="line.2750"></a> -<span class="sourceLineNo">2751</span> boolean success = rss.reportFileArchivalForQuotas(getTableName(), filesWithSizes);<a name="line.2751"></a> -<span class="sourceLineNo">2752</span> if (!success) {<a name="line.2752"></a> -<span class="sourceLineNo">2753</span> LOG.warn("Failed to report archival of files: " + filesWithSizes);<a name="line.2753"></a> -<span class="sourceLineNo">2754</span> }<a name="line.2754"></a> -<span class="sourceLineNo">2755</span> }<a name="line.2755"></a> -<span class="sourceLineNo">2756</span><a name="line.2756"></a> -<span class="sourceLineNo">2757</span> public int getCurrentParallelPutCount() {<a name="line.2757"></a> -<span class="sourceLineNo">2758</span> return currentParallelPutCount.get();<a name="line.2758"></a> +<span class="sourceLineNo">2739</span> return length;<a name="line.2739"></a> +<span class="sourceLineNo">2740</span> }<a name="line.2740"></a> +<span class="sourceLineNo">2741</span><a name="line.2741"></a> +<span class="sourceLineNo">2742</span> public Long preFlushSeqIDEstimation() {<a name="line.2742"></a> +<span class="sourceLineNo">2743</span> return memstore.preFlushSeqIDEstimation();<a name="line.2743"></a> +<span class="sourceLineNo">2744</span> }<a name="line.2744"></a> +<span class="sourceLineNo">2745</span><a name="line.2745"></a> +<span class="sourceLineNo">2746</span> @Override<a name="line.2746"></a> +<span class="sourceLineNo">2747</span> public boolean isSloppyMemStore() {<a name="line.2747"></a> +<span class="sourceLineNo">2748</span> return this.memstore.isSloppy();<a name="line.2748"></a> +<span class="sourceLineNo">2749</span> }<a name="line.2749"></a> +<span class="sourceLineNo">2750</span><a name="line.2750"></a> +<span class="sourceLineNo">2751</span> private void clearCompactedfiles(List<HStoreFile> filesToRemove) throws IOException {<a name="line.2751"></a> +<span class="sourceLineNo">2752</span> LOG.trace("Clearing the compacted file {} from this store", filesToRemove);<a name="line.2752"></a> +<span class="sourceLineNo">2753</span> try {<a name="line.2753"></a> +<span class="sourceLineNo">2754</span> lock.writeLock().lock();<a name="line.2754"></a> +<span class="sourceLineNo">2755</span> this.getStoreEngine().getStoreFileManager().removeCompactedFiles(filesToRemove);<a name="line.2755"></a> +<span class="sourceLineNo">2756</span> } finally {<a name="line.2756"></a> +<span class="sourceLineNo">2757</span> lock.writeLock().unlock();<a name="line.2757"></a> +<span class="sourceLineNo">2758</span> }<a name="line.2758"></a> <span class="sourceLineNo">2759</span> }<a name="line.2759"></a> <span class="sourceLineNo">2760</span><a name="line.2760"></a> -<span class="sourceLineNo">2761</span>}<a name="line.2761"></a> +<span class="sourceLineNo">2761</span> void reportArchivedFilesForQuota(List<? extends StoreFile> archivedFiles, List<Long> fileSizes) {<a name="line.2761"></a> +<span class="sourceLineNo">2762</span> // Sanity check from the caller<a name="line.2762"></a> +<span class="sourceLineNo">2763</span> if (archivedFiles.size() != fileSizes.size()) {<a name="line.2763"></a> +<span class="sourceLineNo">2764</span> throw new RuntimeException("Coding error: should never see lists of varying size");<a name="line.2764"></a> +<span class="sourceLineNo">2765</span> }<a name="line.2765"></a> +<span class="sourceLineNo">2766</span> RegionServerServices rss = this.region.getRegionServerServices();<a name="line.2766"></a> +<span class="sourceLineNo">2767</span> if (rss == null) {<a name="line.2767"></a> +<span class="sourceLineNo">2768</span> return;<a name="line.2768"></a> +<span class="sourceLineNo">2769</span> }<a name="line.2769"></a> +<span class="sourceLineNo">2770</span> List<Entry<String,Long>> filesWithSizes = new ArrayList<>(archivedFiles.size());<a name="line.2770"></a> +<span class="sourceLineNo">2771</span> Iterator<Long> fileSizeIter = fileSizes.iterator();<a name="line.2771"></a> +<span class="sourceLineNo">2772</span> for (StoreFile storeFile : archivedFiles) {<a name="line.2772"></a> +<span class="sourceLineNo">2773</span> final long fileSize = fileSizeIter.next();<a name="line.2773"></a> +<span class="sourceLineNo">2774</span> if (storeFile.isHFile() && fileSize != 0) {<a name="line.2774"></a> +<span class="sourceLineNo">2775</span> filesWithSizes.add(Maps.immutableEntry(storeFile.getPath().getName(), fileSize));<a name="line.2775"></a> +<span class="sourceLineNo">2776</span> }<a name="line.2776"></a> +<span class="sourceLineNo">2777</span> }<a name="line.2777"></a> +<span class="sourceLineNo">2778</span> if (LOG.isTraceEnabled()) {<a name="line.2778"></a> +<span class="sourceLineNo">2779</span> LOG.trace("Files archived: " + archivedFiles + ", reporting the following to the Master: "<a name="line.2779"></a> +<span class="sourceLineNo">2780</span> + filesWithSizes);<a name="line.2780"></a> +<span class="sourceLineNo">2781</span> }<a name="line.2781"></a> +<span class="sourceLineNo">2782</span> boolean success = rss.reportFileArchivalForQuotas(getTableName(), filesWithSizes);<a name="line.2782"></a> +<span class="sourceLineNo">2783</span> if (!success) {<a name="line.2783"></a> +<span class="sourceLineNo">2784</span> LOG.warn("Failed to report archival of files: " + filesWithSizes);<a name="line.2784"></a> +<span class="sourceLineNo">2785</span> }<a name="line.2785"></a> +<span class="sourceLineNo">2786</span> }<a name="line.2786"></a> +<span class="sourceLineNo">2787</span><a name="line.2787"></a> +<span class="sourceLineNo">2788</span> public int getCurrentParallelPutCount() {<a name="line.2788"></a> +<span class="sourceLineNo">2789</span> return currentParallelPutCount.get();<a name="line.2789"></a> +<span class="sourceLineNo">2790</span> }<a name="line.2790"></a> +<span class="sourceLineNo">2791</span><a name="line.2791"></a> +<span class="sourceLineNo">2792</span>}<a name="line.2792"></a>
