public RegionReplicaSinkWriter(RegionReplicaOutputSink sink, ClusterConnection connection, ExecutorService pool, int operationTimeout) { this.sink = sink; this.connection = connection; this.operationTimeout = operationTimeout; this.rpcRetryingCallerFactory = RpcRetryingCallerFactory.instantiate(connection.getConfiguration()); this.rpcControllerFactory = RpcControllerFactory.instantiate(connection.getConfiguration()); this.pool = pool; int nonExistentTableCacheExpiryMs = connection.getConfiguration() .getInt("hbase.region.replica.replication.cache.disabledAndDroppedTables.expiryMs", 5000); // A cache for non existing tables that have a default expiry of 5 sec. This means that if the // table is created again with the same name, we might miss to replicate for that amount of // time. But this cache prevents overloading meta requests for every edit from a deleted file. disabledAndDroppedTables = CacheBuilder.newBuilder() .expireAfterWrite(nonExistentTableCacheExpiryMs, TimeUnit.MILLISECONDS) .initialCapacity(10) .maximumSize(1000) .build(); }
private void replayEdits(final HRegionLocation regionLoc, final HRegionInfo regionInfo, final List<Entry> entries) throws IOException { try { RpcRetryingCallerFactory factory = RpcRetryingCallerFactory.instantiate(conf, null); ReplayServerCallable<ReplicateWALEntryResponse> callable = new ReplayServerCallable<ReplicateWALEntryResponse>(this.conn, this.tableName, regionLoc, regionInfo, entries); factory.<ReplicateWALEntryResponse> newCaller().callWithRetries(callable, this.replayTimeout); } catch (IOException ie) { if (skipErrors) { LOG.warn(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS + "=true so continuing replayEdits with error:" + ie.getMessage()); } else { throw ie; } } }
private void replicateUsingCallable(ClusterConnection connection, Queue<Entry> entries) throws IOException, RuntimeException { Entry entry; while ((entry = entries.poll()) != null) { byte[] row = entry.getEdit().getCells().get(0).getRow(); RegionLocations locations = connection.locateRegion(tableName, row, true, true); RegionReplicaReplayCallable callable = new RegionReplicaReplayCallable(connection, RpcControllerFactory.instantiate(connection.getConfiguration()), table.getName(), locations.getRegionLocation(1), locations.getRegionLocation(1).getRegionInfo(), row, Lists.newArrayList(entry), new AtomicLong()); RpcRetryingCallerFactory factory = RpcRetryingCallerFactory.instantiate( connection.getConfiguration()); factory.<ReplicateWALEntryResponse> newCaller().callWithRetries(callable, 10000); } }
private void replayEdits(final HRegionLocation regionLoc, final HRegionInfo regionInfo, final List<HLog.Entry> entries) throws IOException { try { RpcRetryingCallerFactory factory = RpcRetryingCallerFactory.instantiate(conf); ReplayServerCallable<ReplicateWALEntryResponse> callable = new ReplayServerCallable<ReplicateWALEntryResponse>(this.conn, this.tableName, regionLoc, regionInfo, entries); factory.<ReplicateWALEntryResponse> newCaller().callWithRetries(callable, this.replayTimeout); } catch (IOException ie) { if (skipErrors) { LOG.warn(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS + "=true so continuing replayEdits with error:" + ie.getMessage()); } else { throw ie; } } }
/** * Attempts to do an atomic load of many hfiles into a region. If it fails, it returns a list of * hfiles that need to be retried. If it is successful it will return an empty list. * <p> * NOTE: To maintain row atomicity guarantees, region server callable should succeed atomically * and fails atomically. * <p> * Protected for testing. * @return empty list if success, list of items to retry on recoverable failure */ @VisibleForTesting protected List<LoadQueueItem> tryAtomicRegionLoad(ClientServiceCallable<byte[]> serviceCallable, final TableName tableName, final byte[] first, final Collection<LoadQueueItem> lqis) throws IOException { try { List<LoadQueueItem> toRetry = new ArrayList<>(); Configuration conf = getConf(); byte[] region = RpcRetryingCallerFactory.instantiate(conf, null).<byte[]> newCaller() .callWithRetries(serviceCallable, Integer.MAX_VALUE); if (region == null) { LOG.warn("Attempt to bulk load region containing " + Bytes.toStringBinary(first) + " into table " + tableName + " with files " + lqis + " failed. This is recoverable and they will be retried."); toRetry.addAll(lqis); // return lqi's to retry } // success return toRetry; } catch (IOException e) { LOG.error("Encountered unrecoverable error from region server, additional details: " + serviceCallable.getExceptionMessageAdditionalDetail(), e); throw e; } }
private void replicateUsingCallable(ClusterConnection connection, Queue<Entry> entries) throws IOException, RuntimeException { Entry entry; while ((entry = entries.poll()) != null) { byte[] row = CellUtil.cloneRow(entry.getEdit().getCells().get(0)); RegionLocations locations = connection.locateRegion(tableName, row, true, true); RegionReplicaReplayCallable callable = new RegionReplicaReplayCallable(connection, RpcControllerFactory.instantiate(connection.getConfiguration()), table.getName(), locations.getRegionLocation(1), locations.getRegionLocation(1).getRegionInfo(), row, Lists.newArrayList(entry), new AtomicLong()); RpcRetryingCallerFactory factory = RpcRetryingCallerFactory.instantiate( connection.getConfiguration()); factory.<ReplicateWALEntryResponse> newCaller().callWithRetries(callable, 10000); } }
@Test(timeout=120000) public void testNoBulkLoadsWithNoWrites() throws Exception { Put p = new Put(Bytes.toBytes("to_reject")); p.addColumn( Bytes.toBytes(SpaceQuotaHelperForTests.F1), Bytes.toBytes("to"), Bytes.toBytes("reject")); TableName tableName = writeUntilViolationAndVerifyViolation(SpaceViolationPolicy.NO_WRITES, p); // The table is now in violation. Try to do a bulk load ClientServiceCallable<Void> callable = generateFileToLoad(tableName, 1, 50); RpcRetryingCallerFactory factory = new RpcRetryingCallerFactory(TEST_UTIL.getConfiguration()); RpcRetryingCaller<Void> caller = factory.<Void> newCaller(); try { caller.callWithRetries(callable, Integer.MAX_VALUE); fail("Expected the bulk load call to fail!"); } catch (SpaceLimitingException e) { // Pass LOG.trace("Caught expected exception", e); } }
public RegionReplicaFlushHandler(Server server, ClusterConnection connection, RpcRetryingCallerFactory rpcRetryingCallerFactory, RpcControllerFactory rpcControllerFactory, int operationTimeout, HRegion region) { super(server, EventType.RS_REGION_REPLICA_FLUSH); this.connection = connection; this.rpcRetryingCallerFactory = rpcRetryingCallerFactory; this.rpcControllerFactory = rpcControllerFactory; this.operationTimeout = operationTimeout; this.region = region; }
public RegionCoprocessorRpcChannel(HConnection conn, TableName table, byte[] row) { this.connection = conn; this.table = table; this.row = row; this.rpcFactory = RpcRetryingCallerFactory.instantiate(conn.getConfiguration(), null); this.operationTimeout = conn.getConfiguration().getInt( HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT); }
public RegionCoprocessorRpcChannel(HConnection conn, TableName table, byte[] row, RpcRetryingCallerFactory rpcFactory, RpcControllerFactory rpcControllerFactory) { this.connection = conn; this.table = table; this.row = row; this.rpcFactory = rpcFactory; this.rpcController = rpcControllerFactory;// RpcRetryingCallerFactory.instantiate(conn.getConfiguration()); }
public RegionCoprocessorRpcChannel(HConnection conn, TableName table, byte[] row) { this.connection = conn; this.table = table; this.row = row; this.rpcFactory = RpcRetryingCallerFactory.instantiate(conn.getConfiguration()); this.operationTimeout = conn.getConfiguration().getInt( HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT); }
public RetryingRpcCallable(RpcRetryingCallerFactory factory, RetryingCallable<V> callable, int timeout) { this.factory = factory; this.callable = callable; this.timeout = timeout; }
public void doAnAction() throws Exception { long iteration = numBulkLoads.getAndIncrement(); Path dir = UTIL.getDataTestDirOnTestFS(String.format("bulkLoad_%08d", iteration)); // create HFiles for different column families FileSystem fs = UTIL.getTestFileSystem(); byte[] val = Bytes.toBytes(String.format("%010d", iteration)); final List<Pair<byte[], String>> famPaths = new ArrayList<Pair<byte[], String>>( NUM_CFS); for (int i = 0; i < NUM_CFS; i++) { Path hfile = new Path(dir, family(i)); byte[] fam = Bytes.toBytes(family(i)); createHFile(fs, hfile, fam, QUAL, val, 1000); famPaths.add(new Pair<byte[], String>(fam, hfile.toString())); } // bulk load HFiles final HConnection conn = UTIL.getHBaseAdmin().getConnection(); RegionServerCallable<Void> callable = new RegionServerCallable<Void>(conn, tableName, Bytes.toBytes("aaa")) { @Override public Void call(int callTimeout) throws Exception { LOG.debug("Going to connect to server " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); byte[] regionName = getLocation().getRegionInfo().getRegionName(); BulkLoadHFileRequest request = RequestConverter.buildBulkLoadHFileRequest(famPaths, regionName, true); getStub().bulkLoadHFile(null, request); return null; } }; RpcRetryingCallerFactory factory = new RpcRetryingCallerFactory(conf); RpcRetryingCaller<Void> caller = factory.<Void> newCaller(); caller.callWithRetries(callable, Integer.MAX_VALUE); // Periodically do compaction to reduce the number of open file handles. if (numBulkLoads.get() % 10 == 0) { // 10 * 50 = 500 open file handles! callable = new RegionServerCallable<Void>(conn, tableName, Bytes.toBytes("aaa")) { @Override public Void call(int callTimeout) throws Exception { LOG.debug("compacting " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); AdminProtos.AdminService.BlockingInterface server = conn.getAdmin(getLocation().getServerName()); CompactRegionRequest request = RequestConverter.buildCompactRegionRequest( getLocation().getRegionInfo().getRegionName(), true, null); server.compactRegion(null, request); numCompactions.incrementAndGet(); return null; } }; caller.callWithRetries(callable, Integer.MAX_VALUE); } }
/** * Create a new ClientScanner for the specified table Note that the passed {@link Scan}'s start * row maybe changed changed. * @param conf The {@link Configuration} to use. * @param scan {@link Scan} to use in this scanner * @param tableName The table that we wish to scan * @param connection Connection identifying the cluster * @throws IOException */ public ClientScanner(final Configuration conf, final Scan scan, final TableName tableName, ClusterConnection connection, RpcRetryingCallerFactory rpcFactory, RpcControllerFactory controllerFactory, ExecutorService pool, int primaryOperationTimeout) throws IOException { if (LOG.isTraceEnabled()) { LOG.trace("Scan table=" + tableName + ", startRow=" + Bytes.toStringBinary(scan.getStartRow())); } this.scan = scan; this.tableName = tableName; this.lastNext = System.currentTimeMillis(); this.connection = connection; this.pool = pool; this.primaryOperationTimeout = primaryOperationTimeout; this.retries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER); if (scan.getMaxResultSize() > 0) { this.maxScannerResultSize = scan.getMaxResultSize(); } else { this.maxScannerResultSize = conf.getLong( HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY, HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE); } this.scannerTimeout = HBaseConfiguration.getInt(conf, HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY, HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD); // check if application wants to collect scan metrics initScanMetrics(scan); // Use the caching from the Scan. If not set, use the default cache setting for this table. if (this.scan.getCaching() > 0) { this.caching = this.scan.getCaching(); } else { this.caching = conf.getInt( HConstants.HBASE_CLIENT_SCANNER_CACHING, HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING); } this.caller = rpcFactory.<Result[]> newCaller(); this.rpcControllerFactory = controllerFactory; this.conf = conf; initializeScannerInConstruction(); }
public void doAnAction() throws Exception { long iteration = numBulkLoads.getAndIncrement(); Path dir = UTIL.getDataTestDirOnTestFS(String.format("bulkLoad_%08d", iteration)); // create HFiles for different column families FileSystem fs = UTIL.getTestFileSystem(); byte[] val = Bytes.toBytes(String.format("%010d", iteration)); final List<Pair<byte[], String>> famPaths = new ArrayList<Pair<byte[], String>>( NUM_CFS); for (int i = 0; i < NUM_CFS; i++) { Path hfile = new Path(dir, family(i)); byte[] fam = Bytes.toBytes(family(i)); createHFile(fs, hfile, fam, QUAL, val, 1000); famPaths.add(new Pair<byte[], String>(fam, hfile.toString())); } // bulk load HFiles final HConnection conn = UTIL.getHBaseAdmin().getConnection(); TableName tbl = TableName.valueOf(tableName); RegionServerCallable<Void> callable = new RegionServerCallable<Void>(conn, tbl, Bytes.toBytes("aaa")) { @Override public Void call() throws Exception { LOG.debug("Going to connect to server " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); byte[] regionName = getLocation().getRegionInfo().getRegionName(); BulkLoadHFileRequest request = RequestConverter.buildBulkLoadHFileRequest(famPaths, regionName, true); getStub().bulkLoadHFile(null, request); return null; } }; RpcRetryingCallerFactory factory = new RpcRetryingCallerFactory(conf); RpcRetryingCaller<Void> caller = factory.<Void> newCaller(); caller.callWithRetries(callable); // Periodically do compaction to reduce the number of open file handles. if (numBulkLoads.get() % 10 == 0) { // 10 * 50 = 500 open file handles! callable = new RegionServerCallable<Void>(conn, tbl, Bytes.toBytes("aaa")) { @Override public Void call() throws Exception { LOG.debug("compacting " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); AdminProtos.AdminService.BlockingInterface server = conn.getAdmin(getLocation().getServerName()); CompactRegionRequest request = RequestConverter.buildCompactRegionRequest( getLocation().getRegionInfo().getRegionName(), true, null); server.compactRegion(null, request); numCompactions.incrementAndGet(); return null; } }; caller.callWithRetries(callable); } }
public void doAnAction() throws Exception { long iteration = numBulkLoads.getAndIncrement(); Path dir = UTIL.getDataTestDirOnTestFS(String.format("bulkLoad_%08d", iteration)); // create HFiles for different column families FileSystem fs = UTIL.getTestFileSystem(); byte[] val = Bytes.toBytes(String.format("%010d", iteration)); final List<Pair<byte[], String>> famPaths = new ArrayList<>(NUM_CFS); for (int i = 0; i < NUM_CFS; i++) { Path hfile = new Path(dir, family(i)); byte[] fam = Bytes.toBytes(family(i)); createHFile(fs, hfile, fam, QUAL, val, 1000); famPaths.add(new Pair<>(fam, hfile.toString())); } // bulk load HFiles final ClusterConnection conn = (ClusterConnection) UTIL.getAdmin().getConnection(); Table table = conn.getTable(tableName); final String bulkToken = new SecureBulkLoadEndpointClient(table).prepareBulkLoad(tableName); RpcControllerFactory rpcControllerFactory = new RpcControllerFactory(UTIL.getConfiguration()); ClientServiceCallable<Void> callable = new ClientServiceCallable<Void>(conn, tableName, Bytes.toBytes("aaa"), rpcControllerFactory.newController(), HConstants.PRIORITY_UNSET) { @Override protected Void rpcCall() throws Exception { LOG.debug("Going to connect to server " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); try (Table table = conn.getTable(getTableName())) { boolean loaded = new SecureBulkLoadEndpointClient(table).bulkLoadHFiles(famPaths, null, bulkToken, getLocation().getRegionInfo().getStartKey()); } return null; } }; RpcRetryingCallerFactory factory = new RpcRetryingCallerFactory(conf); RpcRetryingCaller<Void> caller = factory.<Void> newCaller(); caller.callWithRetries(callable, Integer.MAX_VALUE); // Periodically do compaction to reduce the number of open file handles. if (numBulkLoads.get() % 5 == 0) { // 5 * 50 = 250 open file handles! callable = new ClientServiceCallable<Void>(conn, tableName, Bytes.toBytes("aaa"), rpcControllerFactory.newController(), HConstants.PRIORITY_UNSET) { @Override protected Void rpcCall() throws Exception { LOG.debug("compacting " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); AdminProtos.AdminService.BlockingInterface server = conn.getAdmin(getLocation().getServerName()); CompactRegionRequest request = RequestConverter.buildCompactRegionRequest( getLocation().getRegionInfo().getRegionName(), true, null); server.compactRegion(null, request); numCompactions.incrementAndGet(); return null; } }; caller.callWithRetries(callable, Integer.MAX_VALUE); } }
@Override public void doAnAction() throws Exception { long iteration = numBulkLoads.getAndIncrement(); Path dir = UTIL.getDataTestDirOnTestFS(String.format("bulkLoad_%08d", iteration)); // create HFiles for different column families FileSystem fs = UTIL.getTestFileSystem(); byte[] val = Bytes.toBytes(String.format("%010d", iteration)); final List<Pair<byte[], String>> famPaths = new ArrayList<>(NUM_CFS); for (int i = 0; i < NUM_CFS; i++) { Path hfile = new Path(dir, family(i)); byte[] fam = Bytes.toBytes(family(i)); createHFile(fs, hfile, fam, QUAL, val, 1000); famPaths.add(new Pair<>(fam, hfile.toString())); } // bulk load HFiles final ClusterConnection conn = (ClusterConnection)UTIL.getConnection(); Table table = conn.getTable(tableName); final String bulkToken = new SecureBulkLoadClient(UTIL.getConfiguration(), table). prepareBulkLoad(conn); ClientServiceCallable<Void> callable = new ClientServiceCallable<Void>(conn, tableName, Bytes.toBytes("aaa"), new RpcControllerFactory(UTIL.getConfiguration()).newController(), HConstants.PRIORITY_UNSET) { @Override public Void rpcCall() throws Exception { LOG.debug("Going to connect to server " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); SecureBulkLoadClient secureClient = null; byte[] regionName = getLocation().getRegionInfo().getRegionName(); try (Table table = conn.getTable(getTableName())) { secureClient = new SecureBulkLoadClient(UTIL.getConfiguration(), table); secureClient.secureBulkLoadHFiles(getStub(), famPaths, regionName, true, null, bulkToken); } return null; } }; RpcRetryingCallerFactory factory = new RpcRetryingCallerFactory(conf); RpcRetryingCaller<Void> caller = factory.<Void> newCaller(); caller.callWithRetries(callable, Integer.MAX_VALUE); // Periodically do compaction to reduce the number of open file handles. if (numBulkLoads.get() % 5 == 0) { // 5 * 50 = 250 open file handles! callable = new ClientServiceCallable<Void>(conn, tableName, Bytes.toBytes("aaa"), new RpcControllerFactory(UTIL.getConfiguration()).newController(), HConstants.PRIORITY_UNSET) { @Override protected Void rpcCall() throws Exception { LOG.debug("compacting " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); AdminProtos.AdminService.BlockingInterface server = conn.getAdmin(getLocation().getServerName()); CompactRegionRequest request = RequestConverter.buildCompactRegionRequest( getLocation().getRegionInfo().getRegionName(), true, null); server.compactRegion(null, request); numCompactions.incrementAndGet(); return null; } }; caller.callWithRetries(callable, Integer.MAX_VALUE); } }
@Override public void doAnAction() throws Exception { long iteration = numBulkLoads.getAndIncrement(); Path dir = UTIL.getDataTestDirOnTestFS(String.format("bulkLoad_%08d", iteration)); // create HFiles for different column families FileSystem fs = UTIL.getTestFileSystem(); byte[] val = Bytes.toBytes(String.format("%010d", iteration)); final List<Pair<byte[], String>> famPaths = new ArrayList<>(NUM_CFS); for (int i = 0; i < NUM_CFS; i++) { Path hfile = new Path(dir, family(i)); byte[] fam = Bytes.toBytes(family(i)); createHFile(fs, hfile, fam, QUAL, val, 1000); famPaths.add(new Pair<>(fam, hfile.toString())); } // bulk load HFiles final ClusterConnection conn = (ClusterConnection) UTIL.getAdmin().getConnection(); RpcControllerFactory rpcControllerFactory = new RpcControllerFactory(UTIL.getConfiguration()); ClientServiceCallable<Void> callable = new ClientServiceCallable<Void>(conn, tableName, Bytes.toBytes("aaa"), rpcControllerFactory.newController(), HConstants.PRIORITY_UNSET) { @Override protected Void rpcCall() throws Exception { LOG.info("Non-secure old client"); byte[] regionName = getLocation().getRegionInfo().getRegionName(); BulkLoadHFileRequest request = RequestConverter .buildBulkLoadHFileRequest(famPaths, regionName, true, null, null); getStub().bulkLoadHFile(null, request); return null; } }; RpcRetryingCallerFactory factory = new RpcRetryingCallerFactory(conf); RpcRetryingCaller<Void> caller = factory.<Void> newCaller(); caller.callWithRetries(callable, Integer.MAX_VALUE); // Periodically do compaction to reduce the number of open file handles. if (numBulkLoads.get() % 5 == 0) { // 5 * 50 = 250 open file handles! callable = new ClientServiceCallable<Void>(conn, tableName, Bytes.toBytes("aaa"), rpcControllerFactory.newController(), HConstants.PRIORITY_UNSET) { @Override protected Void rpcCall() throws Exception { LOG.debug("compacting " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); AdminProtos.AdminService.BlockingInterface server = conn.getAdmin(getLocation().getServerName()); CompactRegionRequest request = RequestConverter.buildCompactRegionRequest( getLocation().getRegionInfo().getRegionName(), true, null); server.compactRegion(null, request); numCompactions.incrementAndGet(); return null; } }; caller.callWithRetries(callable, Integer.MAX_VALUE); } }
public void doAnAction() throws Exception { long iteration = numBulkLoads.getAndIncrement(); Path dir = UTIL.getDataTestDirOnTestFS(String.format("bulkLoad_%08d", iteration)); // create HFiles for different column families FileSystem fs = UTIL.getTestFileSystem(); byte[] val = Bytes.toBytes(String.format("%010d", iteration)); final List<Pair<byte[], String>> famPaths = new ArrayList<Pair<byte[], String>>( NUM_CFS); for (int i = 0; i < NUM_CFS; i++) { Path hfile = new Path(dir, family(i)); byte[] fam = Bytes.toBytes(family(i)); createHFile(fs, hfile, fam, QUAL, val, 1000); famPaths.add(new Pair<byte[], String>(fam, hfile.toString())); } // bulk load HFiles final HConnection conn = UTIL.getHBaseAdmin().getConnection(); TableName tbl = TableName.valueOf(tableName); RegionServerCallable<Void> callable = new RegionServerCallable<Void>(conn, tbl, Bytes.toBytes("aaa")) { @Override public Void call(int callTimeout) throws Exception { LOG.debug("Going to connect to server " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); byte[] regionName = getLocation().getRegionInfo().getRegionName(); BulkLoadHFileRequest request = RequestConverter.buildBulkLoadHFileRequest(famPaths, regionName, true); getStub().bulkLoadHFile(null, request); return null; } }; RpcRetryingCallerFactory factory = new RpcRetryingCallerFactory(conf); RpcRetryingCaller<Void> caller = factory.<Void> newCaller(); caller.callWithRetries(callable, Integer.MAX_VALUE); // Periodically do compaction to reduce the number of open file handles. if (numBulkLoads.get() % 10 == 0) { // 10 * 50 = 500 open file handles! callable = new RegionServerCallable<Void>(conn, tbl, Bytes.toBytes("aaa")) { @Override public Void call(int callTimeout) throws Exception { LOG.debug("compacting " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); AdminProtos.AdminService.BlockingInterface server = conn.getAdmin(getLocation().getServerName()); CompactRegionRequest request = RequestConverter.buildCompactRegionRequest( getLocation().getRegionInfo().getRegionName(), true, null); server.compactRegion(null, request); numCompactions.incrementAndGet(); return null; } }; caller.callWithRetries(callable, Integer.MAX_VALUE); } }
public RegionCoprocessorRpcChannel(HConnection conn, TableName table, byte[] row) { this.connection = conn; this.table = table; this.row = row; this.rpcFactory = RpcRetryingCallerFactory.instantiate(conn.getConfiguration()); }
