001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.util;
019
020import java.io.IOException;
021import java.io.InterruptedIOException;
022import java.util.ArrayList;
023import java.util.Collection;
024import java.util.HashSet;
025import java.util.List;
026import java.util.Objects;
027import java.util.Set;
028import java.util.concurrent.Callable;
029import java.util.concurrent.CompletionService;
030import java.util.concurrent.ExecutionException;
031import java.util.concurrent.ExecutorCompletionService;
032import java.util.concurrent.ThreadPoolExecutor;
033import java.util.concurrent.TimeUnit;
034import org.apache.hadoop.conf.Configuration;
035import org.apache.hadoop.fs.Path;
036import org.apache.hadoop.hbase.DoNotRetryIOException;
037import org.apache.hadoop.hbase.HConstants;
038import org.apache.hadoop.hbase.client.RegionInfo;
039import org.apache.hadoop.hbase.client.RegionInfoBuilder;
040import org.apache.hadoop.hbase.client.TableDescriptor;
041import org.apache.hadoop.hbase.master.assignment.RegionStates;
042import org.apache.hadoop.hbase.regionserver.HRegion;
043import org.apache.yetus.audience.InterfaceAudience;
044import org.slf4j.Logger;
045import org.slf4j.LoggerFactory;
046
047import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
048
049/**
050 * Utility methods for interacting with the regions.
051 */
052@InterfaceAudience.Private
053public abstract class ModifyRegionUtils {
054  private static final Logger LOG = LoggerFactory.getLogger(ModifyRegionUtils.class);
055
056  private ModifyRegionUtils() {
057  }
058
059  public interface RegionFillTask {
060    void fillRegion(final HRegion region) throws IOException;
061  }
062
063  public interface RegionEditTask {
064    void editRegion(final RegionInfo region) throws IOException;
065  }
066
067  public static RegionInfo[] createRegionInfos(TableDescriptor tableDescriptor,
068    byte[][] splitKeys) {
069    long regionId = EnvironmentEdgeManager.currentTime();
070    RegionInfo[] hRegionInfos = null;
071    if (splitKeys == null || splitKeys.length == 0) {
072      hRegionInfos = new RegionInfo[] { RegionInfoBuilder.newBuilder(tableDescriptor.getTableName())
073        .setStartKey(null).setEndKey(null).setSplit(false).setRegionId(regionId).build() };
074    } else {
075      int numRegions = splitKeys.length + 1;
076      hRegionInfos = new RegionInfo[numRegions];
077      byte[] startKey = null;
078      byte[] endKey = null;
079      for (int i = 0; i < numRegions; i++) {
080        endKey = (i == splitKeys.length) ? null : splitKeys[i];
081        hRegionInfos[i] = RegionInfoBuilder.newBuilder(tableDescriptor.getTableName())
082          .setStartKey(startKey).setEndKey(endKey).setSplit(false).setRegionId(regionId).build();
083        startKey = endKey;
084      }
085    }
086    return hRegionInfos;
087  }
088
089  /**
090   * Checks candidate regions for encoded\-name collisions. Ensures there are no duplicates in the
091   * input and no conflicts with existing region states.
092   */
093  public static void checkForEncodedNameCollisions(final Collection<RegionInfo> candidates,
094    final RegionStates regionStates) throws IOException {
095    if (candidates == null || candidates.isEmpty()) {
096      return;
097    }
098    Objects.requireNonNull(regionStates, "regionStates is null");
099    Set<String> candidateNames = new HashSet<>();
100    for (RegionInfo ri : candidates) {
101      String encoded = ri.getEncodedName();
102      if (
103        !candidateNames.add(encoded)
104          || regionStates.getRegionStateNodeFromEncodedRegionName(encoded) != null
105      ) {
106        throw new DoNotRetryIOException("Encoded region name collision detected: '" + encoded
107          + "' for table " + ri.getTable() + ". Refusing to proceed.");
108      }
109    }
110  }
111
112  /**
113   * Create new set of regions on the specified file-system. NOTE: that you should add the regions
114   * to hbase:meta after this operation.
115   * @param conf            {@link Configuration}
116   * @param rootDir         Root directory for HBase instance
117   * @param tableDescriptor description of the table
118   * @param newRegions      {@link RegionInfo} that describes the regions to create
119   * @param task            {@link RegionFillTask} custom code to populate region after creation
120   */
121  public static List<RegionInfo> createRegions(final Configuration conf, final Path rootDir,
122    final TableDescriptor tableDescriptor, final RegionInfo[] newRegions, final RegionFillTask task)
123    throws IOException {
124    if (newRegions == null) return null;
125    int regionNumber = newRegions.length;
126    ThreadPoolExecutor exec = getRegionOpenAndInitThreadPool(conf,
127      "RegionOpenAndInit-" + tableDescriptor.getTableName(), regionNumber);
128    try {
129      return createRegions(exec, conf, rootDir, tableDescriptor, newRegions, task);
130    } finally {
131      exec.shutdownNow();
132    }
133  }
134
135  /**
136   * Create new set of regions on the specified file-system. NOTE: that you should add the regions
137   * to hbase:meta after this operation.
138   * @param exec            Thread Pool Executor
139   * @param conf            {@link Configuration}
140   * @param rootDir         Root directory for HBase instance
141   * @param tableDescriptor description of the table
142   * @param newRegions      {@link RegionInfo} that describes the regions to create
143   * @param task            {@link RegionFillTask} custom code to populate region after creation
144   */
145  public static List<RegionInfo> createRegions(final ThreadPoolExecutor exec,
146    final Configuration conf, final Path rootDir, final TableDescriptor tableDescriptor,
147    final RegionInfo[] newRegions, final RegionFillTask task) throws IOException {
148    if (newRegions == null) return null;
149    int regionNumber = newRegions.length;
150    CompletionService<RegionInfo> completionService = new ExecutorCompletionService<>(exec);
151    List<RegionInfo> regionInfos = new ArrayList<>();
152    for (final RegionInfo newRegion : newRegions) {
153      completionService.submit(new Callable<RegionInfo>() {
154        @Override
155        public RegionInfo call() throws IOException {
156          return createRegion(conf, rootDir, tableDescriptor, newRegion, task);
157        }
158      });
159    }
160    try {
161      // wait for all regions to finish creation
162      for (int i = 0; i < regionNumber; i++) {
163        regionInfos.add(completionService.take().get());
164      }
165    } catch (InterruptedException e) {
166      LOG.error("Caught " + e + " during region creation");
167      throw new InterruptedIOException(e.getMessage());
168    } catch (ExecutionException e) {
169      throw new IOException(e);
170    }
171    return regionInfos;
172  }
173
174  /**
175   * Create new set of regions on the specified file-system.
176   * @param conf            {@link Configuration}
177   * @param rootDir         Root directory for HBase instance
178   * @param tableDescriptor description of the table
179   * @param newRegion       {@link RegionInfo} that describes the region to create
180   * @param task            {@link RegionFillTask} custom code to populate region after creation
181   */
182  public static RegionInfo createRegion(final Configuration conf, final Path rootDir,
183    final TableDescriptor tableDescriptor, final RegionInfo newRegion, final RegionFillTask task)
184    throws IOException {
185    // 1. Create HRegion
186    // The WAL subsystem will use the default rootDir rather than the passed in rootDir
187    // unless I pass along via the conf.
188    Configuration confForWAL = new Configuration(conf);
189    confForWAL.set(HConstants.HBASE_DIR, rootDir.toString());
190    HRegion region = HRegion.createHRegion(newRegion, rootDir, conf, tableDescriptor, null, false);
191    try {
192      // 2. Custom user code to interact with the created region
193      if (task != null) {
194        task.fillRegion(region);
195      }
196    } finally {
197      // 3. Close the new region to flush to disk. Close log file too.
198      region.close(false, true);
199    }
200    return region.getRegionInfo();
201  }
202
203  /**
204   * Execute the task on the specified set of regions.
205   * @param exec    Thread Pool Executor
206   * @param regions {@link RegionInfo} that describes the regions to edit
207   * @param task    {@link RegionFillTask} custom code to edit the region
208   */
209  public static void editRegions(final ThreadPoolExecutor exec,
210    final Collection<RegionInfo> regions, final RegionEditTask task) throws IOException {
211    final ExecutorCompletionService<Void> completionService = new ExecutorCompletionService<>(exec);
212    for (final RegionInfo hri : regions) {
213      completionService.submit(new Callable<Void>() {
214        @Override
215        public Void call() throws IOException {
216          task.editRegion(hri);
217          return null;
218        }
219      });
220    }
221
222    try {
223      for (RegionInfo hri : regions) {
224        completionService.take().get();
225      }
226    } catch (InterruptedException e) {
227      throw new InterruptedIOException(e.getMessage());
228    } catch (ExecutionException e) {
229      throw new IOException(e.getCause());
230    }
231  }
232
233  /*
234   * used by createRegions() to get the thread pool executor based on the
235   * "hbase.hregion.open.and.init.threads.max" property.
236   */
237  static ThreadPoolExecutor getRegionOpenAndInitThreadPool(final Configuration conf,
238    final String threadNamePrefix, int regionNumber) {
239    int maxThreads =
240      Math.min(regionNumber, conf.getInt("hbase.hregion.open.and.init.threads.max", 16));
241    ThreadPoolExecutor regionOpenAndInitThreadPool = Threads.getBoundedCachedThreadPool(maxThreads,
242      30L, TimeUnit.SECONDS, new ThreadFactoryBuilder().setNameFormat(threadNamePrefix + "-pool-%d")
243        .setDaemon(true).setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
244    return regionOpenAndInitThreadPool;
245  }
246}