HBASE-26810 Add dynamic configuration support for system coprocessors (…

…#4181)

Signed-off-by: Ankit Singhal <ankit@apache.org>

hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java

@@ -101,6 +101,7 @@
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.client.TableState;
+import org.apache.hadoop.hbase.conf.ConfigurationManager;
 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
 import org.apache.hadoop.hbase.exceptions.MasterStoppedException;
 import org.apache.hadoop.hbase.executor.ExecutorType;
@@ -220,6 +221,7 @@
 import org.apache.hadoop.hbase.util.Addressing;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.CoprocessorConfigurationUtil;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.FSTableDescriptors;
 import org.apache.hadoop.hbase.util.FutureUtils;
@@ -391,7 +393,7 @@ public class HMaster extends HBaseServerBase<MasterRpcServices> implements Maste
   // the key is table name, the value is the number of compactions in that table.
   private Map<TableName, AtomicInteger> mobCompactionStates = Maps.newConcurrentMap();
 
-  MasterCoprocessorHost cpHost;
+  volatile MasterCoprocessorHost cpHost;
 
   private final boolean preLoadTableDescriptors;
 
@@ -985,14 +987,9 @@ private void finishActiveMasterInitialization(MonitoredTask status) throws IOExc
     zombieDetector.start();
 
     if (!maintenanceMode) {
-      // Add the Observer to delete quotas on table deletion before starting all CPs by
-      // default with quota support, avoiding if user specifically asks to not load this Observer.
-      if (QuotaUtil.isQuotaEnabled(conf)) {
-        updateConfigurationForQuotasObserver(conf);
-      }
-      // initialize master side coprocessors before we start handling requests
       status.setStatus("Initializing master coprocessors");
-      this.cpHost = new MasterCoprocessorHost(this, this.conf);
+      setQuotasObserver(conf);
+      initializeCoprocessorHost(conf);
     } else {
       // start an in process region server for carrying system regions
       maintenanceRegionServer =
@@ -4097,6 +4094,14 @@ public void onConfigurationChange(Configuration newConf) {
     } catch (IOException e) {
       LOG.warn("Failed to initialize SuperUsers on reloading of the configuration");
     }
+    // append the quotas observer back to the master coprocessor key
+    setQuotasObserver(newConf);
+    // update region server coprocessor if the configuration has changed.
+    if (CoprocessorConfigurationUtil.checkConfigurationChange(getConfiguration(), newConf,
+      CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY) && !maintenanceMode) {
+      LOG.info("Update the master coprocessor(s) because the configuration has changed");
+      initializeCoprocessorHost(newConf);
+    }
   }
 
   @Override
@@ -4166,4 +4171,24 @@ static void setDisableBalancerChoreForTest(boolean disable) {
     disableBalancerChoreForTest = disable;
   }
 
+  @RestrictedApi(explanation = "Should only be called in tests", link = "",
+    allowedOnPath = ".*/src/test/.*")
+  public ConfigurationManager getConfigurationManager() {
+    return configurationManager;
+  }
+
+
+  private void setQuotasObserver(Configuration conf) {
+    // Add the Observer to delete quotas on table deletion before starting all CPs by
+    // default with quota support, avoiding if user specifically asks to not load this Observer.
+    if (QuotaUtil.isQuotaEnabled(conf)) {
+      updateConfigurationForQuotasObserver(conf);
+    }
+  }
+
+  private void initializeCoprocessorHost(Configuration conf) {
+    // initialize master side coprocessors before we start handling requests
+    this.cpHost = new MasterCoprocessorHost(this, conf);
+  }
+
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -164,6 +164,7 @@
 import org.apache.hadoop.hbase.util.CancelableProgressable;
 import org.apache.hadoop.hbase.util.ClassSize;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.CoprocessorConfigurationUtil;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.FSUtils;
 import org.apache.hadoop.hbase.util.HashedBytes;
@@ -709,7 +710,7 @@ void sawNoSuchFamily() {
   private final MultiVersionConcurrencyControl mvcc;
 
   // Coprocessor host
-  private RegionCoprocessorHost coprocessorHost;
+  private volatile RegionCoprocessorHost coprocessorHost;
 
   private TableDescriptor htableDescriptor = null;
   private RegionSplitPolicy splitPolicy;
@@ -8602,6 +8603,14 @@ IOException throwOnInterrupt(Throwable t) {
   @Override
   public void onConfigurationChange(Configuration conf) {
     this.storeHotnessProtector.update(conf);
+    // update coprocessorHost if the configuration has changed.
+    if (CoprocessorConfigurationUtil.checkConfigurationChange(getReadOnlyConfiguration(), conf,
+      CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,
+      CoprocessorHost.USER_REGION_COPROCESSOR_CONF_KEY)) {
+      LOG.info("Update the system coprocessors because the configuration has changed");
+      decorateRegionConfiguration(conf);
+      this.coprocessorHost = new RegionCoprocessorHost(this, rsServices, conf);
+    }
   }
 
   /**

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -139,6 +139,7 @@
 import org.apache.hadoop.hbase.security.UserProvider;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CompressionTest;
+import org.apache.hadoop.hbase.util.CoprocessorConfigurationUtil;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.FSUtils;
 import org.apache.hadoop.hbase.util.FutureUtils;
@@ -409,7 +410,7 @@ public class HRegionServer extends HBaseServerBase<RSRpcServices>
   // chore for refreshing store files for secondary regions
   private StorefileRefresherChore storefileRefresher;
 
-  private RegionServerCoprocessorHost rsHost;
+  private volatile RegionServerCoprocessorHost rsHost;
 
   private RegionServerProcedureManagerHost rspmHost;
 
@@ -3342,6 +3343,13 @@ public void onConfigurationChange(Configuration newConf) {
     } catch (IOException e) {
       LOG.warn("Failed to initialize SuperUsers on reloading of the configuration");
     }
+
+    // update region server coprocessor if the configuration has changed.
+    if (CoprocessorConfigurationUtil.checkConfigurationChange(getConfiguration(), newConf,
+      CoprocessorHost.REGIONSERVER_COPROCESSOR_CONF_KEY)) {
+      LOG.info("Update region server coprocessors because the configuration has changed");
+      this.rsHost = new RegionServerCoprocessorHost(this, newConf);
+    }
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/util/CoprocessorConfigurationUtil.java

@@ -0,0 +1,50 @@
+/*
+ * 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.util;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
+
+/**
+ * Helper class for coprocessor host when configuration changes.
+ */
+@InterfaceAudience.Private
+public final class CoprocessorConfigurationUtil {
+
+  private CoprocessorConfigurationUtil() {
+  }
+
+  public static boolean checkConfigurationChange(Configuration oldConfig, Configuration newConfig,
+    String... configurationKey) {
+    Preconditions.checkArgument(configurationKey != null, "Configuration Key(s) must be provided");
+    boolean isConfigurationChange = false;
+    for (String key : configurationKey) {
+      String oldValue = oldConfig.get(key);
+      String newValue = newConfig.get(key);
+      // check if the coprocessor key has any difference
+      if (!StringUtils.equalsIgnoreCase(oldValue, newValue)) {
+        isConfigurationChange = true;
+        break;
+      }
+    }
+    return isConfigurationChange;
+  }
+}

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java

@@ -52,6 +52,7 @@
 import java.util.Map;
 import java.util.NavigableMap;
 import java.util.Objects;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
@@ -119,6 +120,9 @@
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
+import org.apache.hadoop.hbase.coprocessor.MetaTableMetrics;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.RegionObserver;
 import org.apache.hadoop.hbase.exceptions.FailedSanityCheckException;
 import org.apache.hadoop.hbase.filter.BigDecimalComparator;
 import org.apache.hadoop.hbase.filter.BinaryComparator;
@@ -7964,4 +7968,63 @@ public void run() {
     assertFalse("Region lock holder should not have been interrupted", holderInterrupted.get());
   }
 
+  @Test
+  public void testRegionOnCoprocessorsChange() throws IOException {
+    byte[] cf1 = Bytes.toBytes("CF1");
+    byte[][] families = { cf1 };
+
+    Configuration conf = new Configuration(CONF);
+    region = initHRegion(tableName, method, conf, families);
+    assertNull(region.getCoprocessorHost());
+
+    // set and verify the system coprocessors for region and user region
+    Configuration newConf = new Configuration(conf);
+    newConf.set(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,
+      MetaTableMetrics.class.getName());
+    newConf.set(CoprocessorHost.USER_REGION_COPROCESSOR_CONF_KEY,
+      NoOpRegionCoprocessor.class.getName());
+    // trigger configuration change
+    region.onConfigurationChange(newConf);
+    assertTrue(region.getCoprocessorHost() != null);
+    Set<String> coprocessors = region.getCoprocessorHost().getCoprocessors();
+    assertTrue(coprocessors.size() == 2);
+    assertTrue(region.getCoprocessorHost().getCoprocessors()
+      .contains(MetaTableMetrics.class.getSimpleName()));
+    assertTrue(region.getCoprocessorHost().getCoprocessors()
+      .contains(NoOpRegionCoprocessor.class.getSimpleName()));
+
+    // remove region coprocessor and keep only user region coprocessor
+    newConf.unset(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY);
+    region.onConfigurationChange(newConf);
+    assertTrue(region.getCoprocessorHost() != null);
+    coprocessors = region.getCoprocessorHost().getCoprocessors();
+    assertTrue(coprocessors.size() == 1);
+    assertTrue(region.getCoprocessorHost().getCoprocessors()
+      .contains(NoOpRegionCoprocessor.class.getSimpleName()));
+  }
+
+  @Test
+  public void testRegionOnCoprocessorsWithoutChange() throws IOException {
+    byte[] cf1 = Bytes.toBytes("CF1");
+    byte[][] families = { cf1 };
+
+    Configuration conf = new Configuration(CONF);
+    conf.set(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,
+      MetaTableMetrics.class.getCanonicalName());
+    region = initHRegion(tableName, method, conf, families);
+    // region service is null in unit test, we need to load the coprocessor once
+    region.setCoprocessorHost(new RegionCoprocessorHost(region, null, conf));
+    RegionCoprocessor regionCoprocessor = region.getCoprocessorHost()
+      .findCoprocessor(MetaTableMetrics.class.getName());
+
+    // simulate when other configuration may have changed and onConfigurationChange execute once
+    region.onConfigurationChange(conf);
+    RegionCoprocessor regionCoprocessorAfterOnConfigurationChange = region.getCoprocessorHost()
+      .findCoprocessor(MetaTableMetrics.class.getName());
+    assertEquals(regionCoprocessor, regionCoprocessorAfterOnConfigurationChange);
+  }
+
+  public static class NoOpRegionCoprocessor implements RegionCoprocessor, RegionObserver {
+    // a empty region coprocessor class
+  }
 }

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerOnlineConfigChange.java

@@ -20,6 +20,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
@@ -28,6 +29,7 @@
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseTestingUtil;
 import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.JMXListener;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.Waiter;
 import org.apache.hadoop.hbase.client.Admin;
@@ -37,6 +39,8 @@
 import org.apache.hadoop.hbase.client.RegionLocator;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
+import org.apache.hadoop.hbase.master.HMaster;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionConfiguration;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
 import org.apache.hadoop.hbase.util.Bytes;
@@ -70,6 +74,7 @@ public class TestRegionServerOnlineConfigChange {
 
   private static Table t1 = null;
   private static HRegionServer rs1 = null;
+  private static HMaster hMaster = null;
   private static byte[] r1name = null;
   private static Region r1 = null;
 
@@ -102,6 +107,7 @@ public void setUp() throws Exception {
       rs1 = hbaseTestingUtility.getHBaseCluster().getRegionServer(
         hbaseTestingUtility.getHBaseCluster().getServerWith(r1name));
       r1 = rs1.getRegion(r1name);
+      hMaster = hbaseTestingUtility.getHBaseCluster().getMaster();
     }
   }
 
@@ -271,4 +277,24 @@ public void testStoreConfigurationOnlineChange() {
         .getLong(TableDescriptorBuilder.MAX_FILESIZE, -1);
     assertEquals(MAX_FILE_SIZE, actualMaxFileSize);
   }
+
+  @Test
+  public void testCoprocessorConfigurationOnlineChange() {
+    assertNull(rs1.getRegionServerCoprocessorHost().findCoprocessor(JMXListener.class.getName()));
+    conf.set(CoprocessorHost.REGIONSERVER_COPROCESSOR_CONF_KEY, JMXListener.class.getName());
+    rs1.getConfigurationManager().notifyAllObservers(conf);
+    assertNotNull(
+      rs1.getRegionServerCoprocessorHost().findCoprocessor(JMXListener.class.getName()));
+  }
+
+  @Test
+  public void testCoprocessorConfigurationOnlineChangeOnMaster() {
+    assertNull(hMaster.getMasterCoprocessorHost().findCoprocessor(JMXListener.class.getName()));
+    conf.set(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY, JMXListener.class.getName());
+    assertFalse(hMaster.isInMaintenanceMode());
+    hMaster.getConfigurationManager().notifyAllObservers(conf);
+    assertNotNull(
+      hMaster.getMasterCoprocessorHost().findCoprocessor(JMXListener.class.getName()));
+  }
+
 }

src/main/asciidoc/_chapters/configuration.adoc

@@ -1276,6 +1276,10 @@ Here are those configurations:
 | Key
 | hbase.ipc.server.fallback-to-simple-auth-allowed
 | hbase.cleaner.scan.dir.concurrent.size
+| hbase.coprocessor.master.classes
+| hbase.coprocessor.region.classes
+| hbase.coprocessor.regionserver.classes
+| hbase.coprocessor.user.region.classes
 | hbase.regionserver.thread.compaction.large
 | hbase.regionserver.thread.compaction.small
 | hbase.regionserver.thread.split