hbase-server/src/main/java/org/apache/hadoop/hbase/master/normalizer/RegionNormalizerWorker.java - hbase - Git at Google

 /*
  * 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.normalizer;

 import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.conf.ConfigurationManager;
 import org.apache.hadoop.hbase.conf.ConfigurationObserver;
 import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver;
 import org.apache.hadoop.hbase.master.MasterServices;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter;
 import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils;

 /**
  * Consumes normalization request targets ({@link TableName}s) off the
  * {@link RegionNormalizerWorkQueue}, dispatches them to the {@link RegionNormalizer}, and executes
  * the resulting {@link NormalizationPlan}s.
  */
 @InterfaceAudience.Private
 class RegionNormalizerWorker implements PropagatingConfigurationObserver, Runnable {
   public static final String HBASE_TABLE_NORMALIZATION_ENABLED =
     "hbase.table.normalization.enabled";
   private static final Logger LOG = LoggerFactory.getLogger(RegionNormalizerWorker.class);

   static final String RATE_LIMIT_BYTES_PER_SEC_KEY =
     "hbase.normalizer.throughput.max_bytes_per_sec";
   private static final long RATE_UNLIMITED_BYTES = 1_000_000_000_000L; // 1TB/sec

   static final String CUMULATIVE_SIZE_LIMIT_MB_KEY = "hbase.normalizer.plans_size_limit.mb";
   static final long DEFAULT_CUMULATIVE_SIZE_LIMIT_MB = Long.MAX_VALUE;

   private final MasterServices masterServices;
   private final RegionNormalizer regionNormalizer;
   private final RegionNormalizerWorkQueue<TableName> workQueue;
   private final RateLimiter rateLimiter;

   private final long[] skippedCount;
   private final boolean defaultNormalizerTableLevel;
   private long splitPlanCount;
   private long mergePlanCount;
   private final AtomicLong cumulativePlansSizeLimitMb;

   RegionNormalizerWorker(final Configuration configuration, final MasterServices masterServices,
     final RegionNormalizer regionNormalizer, final RegionNormalizerWorkQueue<TableName> workQueue) {
     this.masterServices = masterServices;
     this.regionNormalizer = regionNormalizer;
     this.workQueue = workQueue;
     this.skippedCount = new long[NormalizationPlan.PlanType.values().length];
     this.splitPlanCount = 0;
     this.mergePlanCount = 0;
     this.rateLimiter = loadRateLimiter(configuration);
     this.defaultNormalizerTableLevel = extractDefaultNormalizerValue(configuration);
     this.cumulativePlansSizeLimitMb = new AtomicLong(
       configuration.getLong(CUMULATIVE_SIZE_LIMIT_MB_KEY, DEFAULT_CUMULATIVE_SIZE_LIMIT_MB));
   }

   private boolean extractDefaultNormalizerValue(final Configuration configuration) {
     String s = configuration.get(HBASE_TABLE_NORMALIZATION_ENABLED);
     return Boolean.parseBoolean(s);
   }

   @Override
   public void registerChildren(ConfigurationManager manager) {
     if (regionNormalizer instanceof ConfigurationObserver) {
       final ConfigurationObserver observer = (ConfigurationObserver) regionNormalizer;
       manager.registerObserver(observer);
     }
   }

   @Override
   public void deregisterChildren(ConfigurationManager manager) {
     if (regionNormalizer instanceof ConfigurationObserver) {
       final ConfigurationObserver observer = (ConfigurationObserver) regionNormalizer;
       manager.deregisterObserver(observer);
     }
   }

   private static long logLongConfigurationUpdated(final String key, final long oldValue,
     final long newValue) {
     if (oldValue != newValue) {
       LOG.info("Updated configuration for key '{}' from {} to {}", key, oldValue, newValue);
     }
     return newValue;
   }

   @Override
   public void onConfigurationChange(Configuration conf) {
     rateLimiter.setRate(loadRateLimit(conf));
     cumulativePlansSizeLimitMb.set(
       logLongConfigurationUpdated(CUMULATIVE_SIZE_LIMIT_MB_KEY, cumulativePlansSizeLimitMb.get(),
         conf.getLong(CUMULATIVE_SIZE_LIMIT_MB_KEY, DEFAULT_CUMULATIVE_SIZE_LIMIT_MB)));
   }

   private static RateLimiter loadRateLimiter(final Configuration configuration) {
     return RateLimiter.create(loadRateLimit(configuration));
   }

   private static long loadRateLimit(final Configuration configuration) {
     long rateLimitBytes =
       configuration.getLongBytes(RATE_LIMIT_BYTES_PER_SEC_KEY, RATE_UNLIMITED_BYTES);
     long rateLimitMbs = rateLimitBytes / 1_000_000L;
     if (rateLimitMbs <= 0) {
       LOG.warn("Configured value {}={} is <= 1MB. Falling back to default.",
         RATE_LIMIT_BYTES_PER_SEC_KEY, rateLimitBytes);
       rateLimitBytes = RATE_UNLIMITED_BYTES;
       rateLimitMbs = RATE_UNLIMITED_BYTES / 1_000_000L;
     }
     LOG.info("Normalizer rate limit set to {}",
       rateLimitBytes == RATE_UNLIMITED_BYTES ? "unlimited" : rateLimitMbs + " MB/sec");
     return rateLimitMbs;
   }

   /**
    * @see RegionNormalizerManager#planSkipped(NormalizationPlan.PlanType)
    */
   void planSkipped(NormalizationPlan.PlanType type) {
     synchronized (skippedCount) {
       // updates come here via procedure threads, so synchronize access to this counter.
       skippedCount[type.ordinal()]++;
     }
   }

   /**
    * @see RegionNormalizerManager#getSkippedCount(NormalizationPlan.PlanType)
    */
   long getSkippedCount(NormalizationPlan.PlanType type) {
     return skippedCount[type.ordinal()];
   }

   /**
    * @see RegionNormalizerManager#getSplitPlanCount()
    */
   long getSplitPlanCount() {
     return splitPlanCount;
   }

   /**
    * @see RegionNormalizerManager#getMergePlanCount()
    */
   long getMergePlanCount() {
     return mergePlanCount;
   }

   /**
    * Used in test only. This field is exposed to the test, as opposed to tracking the current
    * configuration value beside the RateLimiter instance and managing synchronization to keep the
    * two in sync.
    */
   RateLimiter getRateLimiter() {
     return rateLimiter;
   }

   @Override
   public void run() {
     while (true) {
       if (Thread.interrupted()) {
         LOG.debug("interrupt detected. terminating.");
         break;
       }
       final TableName tableName;
       try {
         tableName = workQueue.take();
       } catch (InterruptedException e) {
         LOG.debug("interrupt detected. terminating.");
         break;
       }

       final List<NormalizationPlan> plans = calculatePlans(tableName);
       submitPlans(plans);
     }
   }

   private List<NormalizationPlan> calculatePlans(final TableName tableName) {
     if (masterServices.skipRegionManagementAction("region normalizer")) {
       return Collections.emptyList();
     }

     final TableDescriptor tblDesc;
     try {
       tblDesc = masterServices.getTableDescriptors().get(tableName);
       boolean normalizationEnabled;
       if (tblDesc != null) {
         String defined = tblDesc.getValue(TableDescriptorBuilder.NORMALIZATION_ENABLED);
         if (defined != null) {
           normalizationEnabled = tblDesc.isNormalizationEnabled();
         } else {
           normalizationEnabled = this.defaultNormalizerTableLevel;
         }
         if (!normalizationEnabled) {
           LOG.debug("Skipping table {} because normalization is disabled in its table properties "
             + "and normalization is also disabled at table level by default", tableName);
           return Collections.emptyList();
         }
       }
     } catch (IOException e) {
       LOG.debug("Skipping table {} because unable to access its table descriptor.", tableName, e);
       return Collections.emptyList();
     }

     List<NormalizationPlan> plans = regionNormalizer.computePlansForTable(tblDesc);

     plans = truncateForSize(plans);

     if (CollectionUtils.isEmpty(plans)) {
       LOG.debug("No normalization required for table {}.", tableName);
       return Collections.emptyList();
     }
     return plans;
   }

   private List<NormalizationPlan> truncateForSize(List<NormalizationPlan> plans) {
     if (cumulativePlansSizeLimitMb.get() != DEFAULT_CUMULATIVE_SIZE_LIMIT_MB) {
       List<NormalizationPlan> maybeTruncatedPlans = new ArrayList<>(plans.size());
       long totalCumulativeSizeMb = 0;
       long truncatedCumulativeSizeMb = 0;
       for (NormalizationPlan plan : plans) {
         totalCumulativeSizeMb += plan.getPlanSizeMb();
         if (totalCumulativeSizeMb <= cumulativePlansSizeLimitMb.get()) {
           truncatedCumulativeSizeMb += plan.getPlanSizeMb();
           maybeTruncatedPlans.add(plan);
         }
       }
       if (maybeTruncatedPlans.size() != plans.size()) {
         LOG.debug(
           "Truncating list of normalization plans that RegionNormalizerWorker will process "
             + "because of {}. Original list had {} plan(s), new list has {} plan(s). "
             + "Original list covered regions with cumulative size {} mb, "
             + "new list covers regions with cumulative size {} mb.",
           CUMULATIVE_SIZE_LIMIT_MB_KEY, plans.size(), maybeTruncatedPlans.size(),
           totalCumulativeSizeMb, truncatedCumulativeSizeMb);
       }
       return maybeTruncatedPlans;
     } else {
       return plans;
     }
   }

   private void submitPlans(final List<NormalizationPlan> plans) {
     // as of this writing, `plan.submit()` is non-blocking and uses Async Admin APIs to submit
     // task, so there's no artificial rate-limiting of merge/split requests due to this serial loop.
     for (NormalizationPlan plan : plans) {
       switch (plan.getType()) {
         case MERGE: {
           submitMergePlan((MergeNormalizationPlan) plan);
           break;
         }
         case SPLIT: {
           submitSplitPlan((SplitNormalizationPlan) plan);
           break;
         }
         case NONE:
           LOG.debug("Nothing to do for {} with PlanType=NONE. Ignoring.", plan);
           planSkipped(plan.getType());
           break;
         default:
           LOG.warn("Plan {} is of an unrecognized PlanType. Ignoring.", plan);
           planSkipped(plan.getType());
           break;
       }
     }
   }

   /**
    * Interacts with {@link MasterServices} in order to execute a plan.
    */
   private void submitMergePlan(final MergeNormalizationPlan plan) {
     final int totalSizeMb;
     try {
       final long totalSizeMbLong = plan.getNormalizationTargets().stream()
         .mapToLong(NormalizationTarget::getRegionSizeMb).reduce(0, Math::addExact);
       totalSizeMb = Math.toIntExact(totalSizeMbLong);
     } catch (ArithmeticException e) {
       LOG.debug("Sum of merge request size overflows rate limiter data type. {}", plan);
       planSkipped(plan.getType());
       return;
     }

     final RegionInfo[] infos = plan.getNormalizationTargets().stream()
       .map(NormalizationTarget::getRegionInfo).toArray(RegionInfo[]::new);
     final long pid;
     try {
       pid = masterServices.mergeRegions(infos, false, HConstants.NO_NONCE, HConstants.NO_NONCE);
     } catch (IOException e) {
       LOG.info("failed to submit plan {}.", plan, e);
       planSkipped(plan.getType());
       return;
     }
     mergePlanCount++;
     LOG.info("Submitted {} resulting in pid {}", plan, pid);
     final long rateLimitedSecs = Math.round(rateLimiter.acquire(Math.max(1, totalSizeMb)));
     LOG.debug("Rate limiting delayed the worker by {}", Duration.ofSeconds(rateLimitedSecs));
   }

   /**
    * Interacts with {@link MasterServices} in order to execute a plan.
    */
   private void submitSplitPlan(final SplitNormalizationPlan plan) {
     final int totalSizeMb;
     try {
       totalSizeMb = Math.toIntExact(plan.getSplitTarget().getRegionSizeMb());
     } catch (ArithmeticException e) {
       LOG.debug("Split request size overflows rate limiter data type. {}", plan);
       planSkipped(plan.getType());
       return;
     }
     final RegionInfo info = plan.getSplitTarget().getRegionInfo();
     final long rateLimitedSecs = Math.round(rateLimiter.acquire(Math.max(1, totalSizeMb)));
     LOG.debug("Rate limiting delayed this operation by {}", Duration.ofSeconds(rateLimitedSecs));

     final long pid;
     try {
       pid = masterServices.splitRegion(info, null, HConstants.NO_NONCE, HConstants.NO_NONCE);
     } catch (IOException e) {
       LOG.info("failed to submit plan {}.", plan, e);
       planSkipped(plan.getType());
       return;
     }
     splitPlanCount++;
     LOG.info("Submitted {} resulting in pid {}", plan, pid);
   }
 }
	/*
	* 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.normalizer;

	import java.io.IOException;
	import java.time.Duration;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.List;
	import java.util.concurrent.atomic.AtomicLong;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.hbase.HConstants;
	import org.apache.hadoop.hbase.TableName;
	import org.apache.hadoop.hbase.client.RegionInfo;
	import org.apache.hadoop.hbase.client.TableDescriptor;
	import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
	import org.apache.hadoop.hbase.conf.ConfigurationManager;
	import org.apache.hadoop.hbase.conf.ConfigurationObserver;
	import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver;
	import org.apache.hadoop.hbase.master.MasterServices;
	import org.apache.yetus.audience.InterfaceAudience;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter;
	import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils;

	/**
	* Consumes normalization request targets ({@link TableName}s) off the
	* {@link RegionNormalizerWorkQueue}, dispatches them to the {@link RegionNormalizer}, and executes
	* the resulting {@link NormalizationPlan}s.
	*/
	@InterfaceAudience.Private
	class RegionNormalizerWorker implements PropagatingConfigurationObserver, Runnable {
	public static final String HBASE_TABLE_NORMALIZATION_ENABLED =
	"hbase.table.normalization.enabled";
	private static final Logger LOG = LoggerFactory.getLogger(RegionNormalizerWorker.class);

	static final String RATE_LIMIT_BYTES_PER_SEC_KEY =
	"hbase.normalizer.throughput.max_bytes_per_sec";
	private static final long RATE_UNLIMITED_BYTES = 1_000_000_000_000L; // 1TB/sec

	static final String CUMULATIVE_SIZE_LIMIT_MB_KEY = "hbase.normalizer.plans_size_limit.mb";
	static final long DEFAULT_CUMULATIVE_SIZE_LIMIT_MB = Long.MAX_VALUE;

	private final MasterServices masterServices;
	private final RegionNormalizer regionNormalizer;
	private final RegionNormalizerWorkQueue<TableName> workQueue;
	private final RateLimiter rateLimiter;

	private final long[] skippedCount;
	private final boolean defaultNormalizerTableLevel;
	private long splitPlanCount;
	private long mergePlanCount;
	private final AtomicLong cumulativePlansSizeLimitMb;

	RegionNormalizerWorker(final Configuration configuration, final MasterServices masterServices,
	final RegionNormalizer regionNormalizer, final RegionNormalizerWorkQueue<TableName> workQueue) {
	this.masterServices = masterServices;
	this.regionNormalizer = regionNormalizer;
	this.workQueue = workQueue;
	this.skippedCount = new long[NormalizationPlan.PlanType.values().length];
	this.splitPlanCount = 0;
	this.mergePlanCount = 0;
	this.rateLimiter = loadRateLimiter(configuration);
	this.defaultNormalizerTableLevel = extractDefaultNormalizerValue(configuration);
	this.cumulativePlansSizeLimitMb = new AtomicLong(
	configuration.getLong(CUMULATIVE_SIZE_LIMIT_MB_KEY, DEFAULT_CUMULATIVE_SIZE_LIMIT_MB));
	}

	private boolean extractDefaultNormalizerValue(final Configuration configuration) {
	String s = configuration.get(HBASE_TABLE_NORMALIZATION_ENABLED);
	return Boolean.parseBoolean(s);
	}

	@Override
	public void registerChildren(ConfigurationManager manager) {
	if (regionNormalizer instanceof ConfigurationObserver) {
	final ConfigurationObserver observer = (ConfigurationObserver) regionNormalizer;
	manager.registerObserver(observer);
	}
	}

	@Override
	public void deregisterChildren(ConfigurationManager manager) {
	if (regionNormalizer instanceof ConfigurationObserver) {
	final ConfigurationObserver observer = (ConfigurationObserver) regionNormalizer;
	manager.deregisterObserver(observer);
	}
	}

	private static long logLongConfigurationUpdated(final String key, final long oldValue,
	final long newValue) {
	if (oldValue != newValue) {
	LOG.info("Updated configuration for key '{}' from {} to {}", key, oldValue, newValue);
	}
	return newValue;
	}

	@Override
	public void onConfigurationChange(Configuration conf) {
	rateLimiter.setRate(loadRateLimit(conf));
	cumulativePlansSizeLimitMb.set(
	logLongConfigurationUpdated(CUMULATIVE_SIZE_LIMIT_MB_KEY, cumulativePlansSizeLimitMb.get(),
	conf.getLong(CUMULATIVE_SIZE_LIMIT_MB_KEY, DEFAULT_CUMULATIVE_SIZE_LIMIT_MB)));
	}

	private static RateLimiter loadRateLimiter(final Configuration configuration) {
	return RateLimiter.create(loadRateLimit(configuration));
	}

	private static long loadRateLimit(final Configuration configuration) {
	long rateLimitBytes =
	configuration.getLongBytes(RATE_LIMIT_BYTES_PER_SEC_KEY, RATE_UNLIMITED_BYTES);
	long rateLimitMbs = rateLimitBytes / 1_000_000L;
	if (rateLimitMbs <= 0) {
	LOG.warn("Configured value {}={} is <= 1MB. Falling back to default.",
	RATE_LIMIT_BYTES_PER_SEC_KEY, rateLimitBytes);
	rateLimitBytes = RATE_UNLIMITED_BYTES;
	rateLimitMbs = RATE_UNLIMITED_BYTES / 1_000_000L;
	}
	LOG.info("Normalizer rate limit set to {}",
	rateLimitBytes == RATE_UNLIMITED_BYTES ? "unlimited" : rateLimitMbs + " MB/sec");
	return rateLimitMbs;
	}

	/**
	* @see RegionNormalizerManager#planSkipped(NormalizationPlan.PlanType)
	*/
	void planSkipped(NormalizationPlan.PlanType type) {
	synchronized (skippedCount) {
	// updates come here via procedure threads, so synchronize access to this counter.
	skippedCount[type.ordinal()]++;
	}
	}

	/**
	* @see RegionNormalizerManager#getSkippedCount(NormalizationPlan.PlanType)
	*/
	long getSkippedCount(NormalizationPlan.PlanType type) {
	return skippedCount[type.ordinal()];
	}

	/**
	* @see RegionNormalizerManager#getSplitPlanCount()
	*/
	long getSplitPlanCount() {
	return splitPlanCount;
	}

	/**
	* @see RegionNormalizerManager#getMergePlanCount()
	*/
	long getMergePlanCount() {
	return mergePlanCount;
	}

	/**
	* Used in test only. This field is exposed to the test, as opposed to tracking the current
	* configuration value beside the RateLimiter instance and managing synchronization to keep the
	* two in sync.
	*/
	RateLimiter getRateLimiter() {
	return rateLimiter;
	}

	@Override
	public void run() {
	while (true) {
	if (Thread.interrupted()) {
	LOG.debug("interrupt detected. terminating.");
	break;
	}
	final TableName tableName;
	try {
	tableName = workQueue.take();
	} catch (InterruptedException e) {
	LOG.debug("interrupt detected. terminating.");
	break;
	}

	final List<NormalizationPlan> plans = calculatePlans(tableName);
	submitPlans(plans);
	}
	}

	private List<NormalizationPlan> calculatePlans(final TableName tableName) {
	if (masterServices.skipRegionManagementAction("region normalizer")) {
	return Collections.emptyList();
	}

	final TableDescriptor tblDesc;
	try {
	tblDesc = masterServices.getTableDescriptors().get(tableName);
	boolean normalizationEnabled;
	if (tblDesc != null) {
	String defined = tblDesc.getValue(TableDescriptorBuilder.NORMALIZATION_ENABLED);
	if (defined != null) {
	normalizationEnabled = tblDesc.isNormalizationEnabled();
	} else {
	normalizationEnabled = this.defaultNormalizerTableLevel;
	}
	if (!normalizationEnabled) {
	LOG.debug("Skipping table {} because normalization is disabled in its table properties "
	+ "and normalization is also disabled at table level by default", tableName);
	return Collections.emptyList();
	}
	}
	} catch (IOException e) {
	LOG.debug("Skipping table {} because unable to access its table descriptor.", tableName, e);
	return Collections.emptyList();
	}

	List<NormalizationPlan> plans = regionNormalizer.computePlansForTable(tblDesc);

	plans = truncateForSize(plans);

	if (CollectionUtils.isEmpty(plans)) {
	LOG.debug("No normalization required for table {}.", tableName);
	return Collections.emptyList();
	}
	return plans;
	}

	private List<NormalizationPlan> truncateForSize(List<NormalizationPlan> plans) {
	if (cumulativePlansSizeLimitMb.get() != DEFAULT_CUMULATIVE_SIZE_LIMIT_MB) {
	List<NormalizationPlan> maybeTruncatedPlans = new ArrayList<>(plans.size());
	long totalCumulativeSizeMb = 0;
	long truncatedCumulativeSizeMb = 0;
	for (NormalizationPlan plan : plans) {
	totalCumulativeSizeMb += plan.getPlanSizeMb();
	if (totalCumulativeSizeMb <= cumulativePlansSizeLimitMb.get()) {
	truncatedCumulativeSizeMb += plan.getPlanSizeMb();
	maybeTruncatedPlans.add(plan);
	}
	}
	if (maybeTruncatedPlans.size() != plans.size()) {
	LOG.debug(
	"Truncating list of normalization plans that RegionNormalizerWorker will process "
	+ "because of {}. Original list had {} plan(s), new list has {} plan(s). "
	+ "Original list covered regions with cumulative size {} mb, "
	+ "new list covers regions with cumulative size {} mb.",
	CUMULATIVE_SIZE_LIMIT_MB_KEY, plans.size(), maybeTruncatedPlans.size(),
	totalCumulativeSizeMb, truncatedCumulativeSizeMb);
	}
	return maybeTruncatedPlans;
	} else {
	return plans;
	}
	}

	private void submitPlans(final List<NormalizationPlan> plans) {
	// as of this writing, `plan.submit()` is non-blocking and uses Async Admin APIs to submit
	// task, so there's no artificial rate-limiting of merge/split requests due to this serial loop.
	for (NormalizationPlan plan : plans) {
	switch (plan.getType()) {
	case MERGE: {
	submitMergePlan((MergeNormalizationPlan) plan);
	break;
	}
	case SPLIT: {
	submitSplitPlan((SplitNormalizationPlan) plan);
	break;
	}
	case NONE:
	LOG.debug("Nothing to do for {} with PlanType=NONE. Ignoring.", plan);
	planSkipped(plan.getType());
	break;
	default:
	LOG.warn("Plan {} is of an unrecognized PlanType. Ignoring.", plan);
	planSkipped(plan.getType());
	break;
	}
	}
	}

	/**
	* Interacts with {@link MasterServices} in order to execute a plan.
	*/
	private void submitMergePlan(final MergeNormalizationPlan plan) {
	final int totalSizeMb;
	try {
	final long totalSizeMbLong = plan.getNormalizationTargets().stream()
	.mapToLong(NormalizationTarget::getRegionSizeMb).reduce(0, Math::addExact);
	totalSizeMb = Math.toIntExact(totalSizeMbLong);
	} catch (ArithmeticException e) {
	LOG.debug("Sum of merge request size overflows rate limiter data type. {}", plan);
	planSkipped(plan.getType());
	return;
	}

	final RegionInfo[] infos = plan.getNormalizationTargets().stream()
	.map(NormalizationTarget::getRegionInfo).toArray(RegionInfo[]::new);
	final long pid;
	try {
	pid = masterServices.mergeRegions(infos, false, HConstants.NO_NONCE, HConstants.NO_NONCE);
	} catch (IOException e) {
	LOG.info("failed to submit plan {}.", plan, e);
	planSkipped(plan.getType());
	return;
	}
	mergePlanCount++;
	LOG.info("Submitted {} resulting in pid {}", plan, pid);
	final long rateLimitedSecs = Math.round(rateLimiter.acquire(Math.max(1, totalSizeMb)));
	LOG.debug("Rate limiting delayed the worker by {}", Duration.ofSeconds(rateLimitedSecs));
	}

	/**
	* Interacts with {@link MasterServices} in order to execute a plan.
	*/
	private void submitSplitPlan(final SplitNormalizationPlan plan) {
	final int totalSizeMb;
	try {
	totalSizeMb = Math.toIntExact(plan.getSplitTarget().getRegionSizeMb());
	} catch (ArithmeticException e) {
	LOG.debug("Split request size overflows rate limiter data type. {}", plan);
	planSkipped(plan.getType());
	return;
	}
	final RegionInfo info = plan.getSplitTarget().getRegionInfo();
	final long rateLimitedSecs = Math.round(rateLimiter.acquire(Math.max(1, totalSizeMb)));
	LOG.debug("Rate limiting delayed this operation by {}", Duration.ofSeconds(rateLimitedSecs));

	final long pid;
	try {
	pid = masterServices.splitRegion(info, null, HConstants.NO_NONCE, HConstants.NO_NONCE);
	} catch (IOException e) {
	LOG.info("failed to submit plan {}.", plan, e);
	planSkipped(plan.getType());
	return;
	}
	splitPlanCount++;
	LOG.info("Submitted {} resulting in pid {}", plan, pid);
	}
	}