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.master.procedure.MasterProcedureEnv;
043import org.apache.hadoop.hbase.regionserver.HRegion;
044import org.apache.yetus.audience.InterfaceAudience;
045import org.slf4j.Logger;
046import org.slf4j.LoggerFactory;
047
048import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
049
050/**
051 * Utility methods for interacting with the regions.
052 */
053@InterfaceAudience.Private
054public abstract class ModifyRegionUtils {
055  private static final Logger LOG = LoggerFactory.getLogger(ModifyRegionUtils.class);
056
057  private ModifyRegionUtils() {
058  }
059
060  public interface RegionFillTask {
061    void fillRegion(final HRegion region) throws IOException;
062  }
063
064  public interface RegionEditTask {
065    void editRegion(final RegionInfo region) throws IOException;
066  }
067
068  public static RegionInfo[] createRegionInfos(TableDescriptor tableDescriptor,
069    byte[][] splitKeys) {
070    long regionId = EnvironmentEdgeManager.currentTime();
071    RegionInfo[] hRegionInfos = null;
072    if (splitKeys == null || splitKeys.length == 0) {
073      hRegionInfos = new RegionInfo[] { RegionInfoBuilder.newBuilder(tableDescriptor.getTableName())
074        .setStartKey(null).setEndKey(null).setSplit(false).setRegionId(regionId).build() };
075    } else {
076      int numRegions = splitKeys.length + 1;
077      hRegionInfos = new RegionInfo[numRegions];
078      byte[] startKey = null;
079      byte[] endKey = null;
080      for (int i = 0; i < numRegions; i++) {
081        endKey = (i == splitKeys.length) ? null : splitKeys[i];
082        hRegionInfos[i] = RegionInfoBuilder.newBuilder(tableDescriptor.getTableName())
083          .setStartKey(startKey).setEndKey(endKey).setSplit(false).setRegionId(regionId).build();
084        startKey = endKey;
085      }
086    }
087    return hRegionInfos;
088  }
089
090  /**
091   * Checks candidate regions for encoded\-name collisions. Ensures there are no duplicates in the
092   * input and no conflicts with existing region states.
093   */
094  public static void checkForEncodedNameCollisions(final Collection<RegionInfo> candidates,
095    final RegionStates regionStates) throws IOException {
096    if (candidates == null || candidates.isEmpty()) {
097      return;
098    }
099    Objects.requireNonNull(regionStates, "regionStates is null");
100    Set<String> candidateNames = new HashSet<>();
101    for (RegionInfo ri : candidates) {
102      String encoded = ri.getEncodedName();
103      if (
104        !candidateNames.add(encoded)
105          || regionStates.getRegionStateNodeFromEncodedRegionName(encoded) != null
106      ) {
107        throw new DoNotRetryIOException("Encoded region name collision detected: '" + encoded
108          + "' for table " + ri.getTable() + ". Refusing to proceed.");
109      }
110    }
111  }
112
113  /**
114   * Create new set of regions on the specified file-system. NOTE: that you should add the regions
115   * to hbase:meta after this operation.
116   * @param env             {@link MasterProcedureEnv}
117   * @param rootDir         Root directory for HBase instance
118   * @param tableDescriptor description of the table
119   * @param newRegions      {@link RegionInfo} that describes the regions to create
120   * @param task            {@link RegionFillTask} custom code to populate region after creation
121   */
122  public static List<RegionInfo> createRegions(final MasterProcedureEnv env, final Path rootDir,
123    final TableDescriptor tableDescriptor, final RegionInfo[] newRegions, final RegionFillTask task)
124    throws IOException {
125    if (newRegions == null) return null;
126    int regionNumber = newRegions.length;
127    ThreadPoolExecutor exec = getRegionOpenAndInitThreadPool(env.getMasterConfiguration(),
128      "RegionOpenAndInit-" + tableDescriptor.getTableName(), regionNumber);
129    try {
130      return createRegions(exec, env.getMasterConfiguration(), env, rootDir, tableDescriptor,
131        newRegions, task);
132    } finally {
133      exec.shutdownNow();
134    }
135  }
136
137  public static List<RegionInfo> createRegions(final ThreadPoolExecutor exec,
138    final Configuration conf, final Path rootDir, final TableDescriptor tableDescriptor,
139    final RegionInfo[] newRegions, final RegionFillTask task) throws IOException {
140    return createRegions(exec, conf, null, rootDir, tableDescriptor, newRegions, task);
141  }
142
143  /**
144   * Create new set of regions on the specified file-system. NOTE: that you should add the regions
145   * to hbase:meta after this operation.
146   * @param exec            Thread Pool Executor
147   * @param conf            {@link Configuration}
148   * @param rootDir         Root directory for HBase instance
149   * @param tableDescriptor description of the table
150   * @param newRegions      {@link RegionInfo} that describes the regions to create
151   * @param task            {@link RegionFillTask} custom code to populate region after creation
152   */
153  public static List<RegionInfo> createRegions(final ThreadPoolExecutor exec,
154    final Configuration conf, final MasterProcedureEnv env, final Path rootDir,
155    final TableDescriptor tableDescriptor, final RegionInfo[] newRegions, final RegionFillTask task)
156    throws IOException {
157    if (newRegions == null) return null;
158    int regionNumber = newRegions.length;
159    CompletionService<RegionInfo> completionService = new ExecutorCompletionService<>(exec);
160    List<RegionInfo> regionInfos = new ArrayList<>();
161    for (final RegionInfo newRegion : newRegions) {
162      completionService.submit(new Callable<RegionInfo>() {
163        @Override
164        public RegionInfo call() throws IOException {
165          return createRegion(conf, env, rootDir, tableDescriptor, newRegion, task);
166        }
167      });
168    }
169    try {
170      // wait for all regions to finish creation
171      for (int i = 0; i < regionNumber; i++) {
172        regionInfos.add(completionService.take().get());
173      }
174    } catch (InterruptedException e) {
175      LOG.error("Caught " + e + " during region creation");
176      throw new InterruptedIOException(e.getMessage());
177    } catch (ExecutionException e) {
178      throw new IOException(e);
179    }
180    return regionInfos;
181  }
182
183  /**
184   * Create new set of regions on the specified file-system.
185   * @param conf            {@link Configuration}
186   * @param rootDir         Root directory for HBase instance
187   * @param tableDescriptor description of the table
188   * @param newRegion       {@link RegionInfo} that describes the region to create
189   * @param task            {@link RegionFillTask} custom code to populate region after creation
190   */
191  public static RegionInfo createRegion(final Configuration conf, final MasterProcedureEnv env,
192    final Path rootDir, final TableDescriptor tableDescriptor, final RegionInfo newRegion,
193    final RegionFillTask task) throws IOException {
194    // 1. Create HRegion
195    // The WAL subsystem will use the default rootDir rather than the passed in rootDir
196    // unless I pass along via the conf.
197    Configuration confForWAL = new Configuration(conf);
198    confForWAL.set(HConstants.HBASE_DIR, rootDir.toString());
199    HRegion region = HRegion.createHRegion(newRegion, rootDir, conf, tableDescriptor, null, false,
200      null, env == null ? null : env.getMasterServices());
201    try {
202      // 2. Custom user code to interact with the created region
203      if (task != null) {
204        task.fillRegion(region);
205      }
206    } finally {
207      // 3. Close the new region to flush to disk. Close log file too.
208      region.close(false, true);
209    }
210    return region.getRegionInfo();
211  }
212
213  /**
214   * Execute the task on the specified set of regions.
215   * @param exec    Thread Pool Executor
216   * @param regions {@link RegionInfo} that describes the regions to edit
217   * @param task    {@link RegionFillTask} custom code to edit the region
218   */
219  public static void editRegions(final ThreadPoolExecutor exec,
220    final Collection<RegionInfo> regions, final RegionEditTask task) throws IOException {
221    final ExecutorCompletionService<Void> completionService = new ExecutorCompletionService<>(exec);
222    for (final RegionInfo hri : regions) {
223      completionService.submit(new Callable<Void>() {
224        @Override
225        public Void call() throws IOException {
226          task.editRegion(hri);
227          return null;
228        }
229      });
230    }
231
232    try {
233      for (RegionInfo hri : regions) {
234        completionService.take().get();
235      }
236    } catch (InterruptedException e) {
237      throw new InterruptedIOException(e.getMessage());
238    } catch (ExecutionException e) {
239      throw new IOException(e.getCause());
240    }
241  }
242
243  /*
244   * used by createRegions() to get the thread pool executor based on the
245   * "hbase.hregion.open.and.init.threads.max" property.
246   */
247  static ThreadPoolExecutor getRegionOpenAndInitThreadPool(final Configuration conf,
248    final String threadNamePrefix, int regionNumber) {
249    int maxThreads =
250      Math.min(regionNumber, conf.getInt("hbase.hregion.open.and.init.threads.max", 16));
251    ThreadPoolExecutor regionOpenAndInitThreadPool = Threads.getBoundedCachedThreadPool(maxThreads,
252      30L, TimeUnit.SECONDS, new ThreadFactoryBuilder().setNameFormat(threadNamePrefix + "-pool-%d")
253        .setDaemon(true).setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());
254    return regionOpenAndInitThreadPool;
255  }
256}