Skip to content

Commit

Permalink
Revert "Revert "HBASE-27062 ThreadPool is unnecessary in HBaseInterCl…
Browse files Browse the repository at this point in the history 
…usterReplication… (#4463)" (#4559)"

This reverts commit 902bc3e.
  • Loading branch information
@comnetwork
comnetwork authored Jun 22, 2022
1 parent 902bc3e commit 1e4fcac
Show file tree
Hide file tree
Showing 6 changed files with 154 additions and 136 deletions.
11 changes: 6 additions & 5 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtobufUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,14 +21,14 @@
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.PrivateCellUtil;
import org.apache.hadoop.hbase.client.AsyncRegionServerAdmin;
import org.apache.hadoop.hbase.io.SizedCellScanner;
import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec;
import org.apache.hadoop.hbase.util.FutureUtils;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.hadoop.hbase.wal.WALEdit;
Expand All @@ -37,6 +37,7 @@
import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;

import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ReplicateWALEntryRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ReplicateWALEntryResponse;
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WALEntry;
import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;

Expand All @@ -52,12 +53,12 @@ public class ReplicationProtobufUtil {
* @param sourceBaseNamespaceDir Path to source cluster base namespace directory
* @param sourceHFileArchiveDir Path to the source cluster hfile archive directory
*/
public static void replicateWALEntry(AsyncRegionServerAdmin admin, Entry[] entries,
String replicationClusterId, Path sourceBaseNamespaceDir, Path sourceHFileArchiveDir,
int timeout) throws IOException {
public static CompletableFuture<ReplicateWALEntryResponse> replicateWALEntry(
AsyncRegionServerAdmin admin, Entry[] entries, String replicationClusterId,
Path sourceBaseNamespaceDir, Path sourceHFileArchiveDir, int timeout) {
Pair<ReplicateWALEntryRequest, CellScanner> p = buildReplicateWALEntryRequest(entries, null,
replicationClusterId, sourceBaseNamespaceDir, sourceHFileArchiveDir);
FutureUtils.get(admin.replicateWALEntry(p.getFirst(), p.getSecond(), timeout));
return admin.replicateWALEntry(p.getFirst(), p.getSecond(), timeout);
}

/**
Expand Down
211 changes: 112 additions & 99 deletions ...rg/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,18 +29,11 @@
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Abortable;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
Expand All @@ -57,15 +50,16 @@
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.util.FutureUtils;
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.hadoop.hbase.wal.WALEdit;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hbase.thirdparty.com.google.common.collect.Iterators;
import org.apache.hbase.thirdparty.com.google.common.collect.PeekingIterator;

/**
* A {@link org.apache.hadoop.hbase.replication.ReplicationEndpoint} implementation for replicating
Expand All @@ -82,8 +76,6 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi
private static final Logger LOG =
LoggerFactory.getLogger(HBaseInterClusterReplicationEndpoint.class);

private static final long DEFAULT_MAX_TERMINATION_WAIT_MULTIPLIER = 2;

/** Drop edits for tables that been deleted from the replication source and target */
public static final String REPLICATION_DROP_ON_DELETED_TABLE_KEY =
"hbase.replication.drop.on.deleted.table";
Expand All @@ -97,25 +89,22 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi
private int maxRetriesMultiplier;
// Socket timeouts require even bolder actions since we don't want to DDOS
private int socketTimeoutMultiplier;
// Amount of time for shutdown to wait for all tasks to complete
private long maxTerminationWait;
// Size limit for replication RPCs, in bytes
private int replicationRpcLimit;
// Metrics for this source
private MetricsSource metrics;
private boolean peersSelected = false;
private String replicationClusterId = "";
private ThreadPoolExecutor exec;
private int maxThreads;
private Path baseNamespaceDir;
private Path hfileArchiveDir;
private boolean replicationBulkLoadDataEnabled;
private Abortable abortable;
private boolean dropOnDeletedTables;
private boolean dropOnDeletedColumnFamilies;
private boolean isSerial = false;
// Initialising as 0 to guarantee at least one logging message
private long lastSinkFetchTime = 0;
private volatile boolean stopping = false;

@Override
public void init(Context context) throws IOException {
Expand All @@ -124,20 +113,11 @@ public void init(Context context) throws IOException {
this.maxRetriesMultiplier = this.conf.getInt("replication.source.maxretriesmultiplier", 300);
this.socketTimeoutMultiplier =
this.conf.getInt("replication.source.socketTimeoutMultiplier", maxRetriesMultiplier);
// A Replicator job is bound by the RPC timeout. We will wait this long for all Replicator
// tasks to terminate when doStop() is called.
long maxTerminationWaitMultiplier = this.conf.getLong(
"replication.source.maxterminationmultiplier", DEFAULT_MAX_TERMINATION_WAIT_MULTIPLIER);
this.maxTerminationWait = maxTerminationWaitMultiplier
* this.conf.getLong(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
this.sleepForRetries = this.conf.getLong("replication.source.sleepforretries", 1000);
this.metrics = context.getMetrics();
// per sink thread pool
this.maxThreads = this.conf.getInt(HConstants.REPLICATION_SOURCE_MAXTHREADS_KEY,
HConstants.REPLICATION_SOURCE_MAXTHREADS_DEFAULT);
this.exec = Threads.getBoundedCachedThreadPool(maxThreads, 60, TimeUnit.SECONDS,
new ThreadFactoryBuilder().setDaemon(true).setNameFormat("SinkThread-%d").build());
this.abortable = ctx.getAbortable();
// Set the size limit for replication RPCs to 95% of the max request size.
// We could do with less slop if we have an accurate estimate of encoded size. Being
// conservative for now.
Expand Down Expand Up @@ -394,43 +374,42 @@ List<List<Entry>> filterNotExistColumnFamilyEdits(final List<List<Entry>> oldEnt
return entryList;
}

private long parallelReplicate(CompletionService<Integer> pool, ReplicateContext replicateContext,
List<List<Entry>> batches) throws IOException {
int futures = 0;
private long parallelReplicate(ReplicateContext replicateContext, List<List<Entry>> batches)
throws IOException {
List<CompletableFuture<Integer>> futures =
new ArrayList<CompletableFuture<Integer>>(batches.size());
for (int i = 0; i < batches.size(); i++) {
List<Entry> entries = batches.get(i);
if (!entries.isEmpty()) {
if (LOG.isTraceEnabled()) {
LOG.trace("{} Submitting {} entries of total size {}", logPeerId(), entries.size(),
replicateContext.getSize());
}
// RuntimeExceptions encountered here bubble up and are handled in ReplicationSource
pool.submit(createReplicator(entries, i, replicateContext.getTimeout()));
futures++;
if (entries.isEmpty()) {
continue;
}
if (LOG.isTraceEnabled()) {
LOG.trace("{} Submitting {} entries of total size {}", logPeerId(), entries.size(),
replicateContext.getSize());
}
// RuntimeExceptions encountered here bubble up and are handled in ReplicationSource
futures.add(asyncReplicate(entries, i, replicateContext.getTimeout()));
}

IOException iox = null;
long lastWriteTime = 0;
for (int i = 0; i < futures; i++) {

for (CompletableFuture<Integer> f : futures) {
try {
// wait for all futures, remove successful parts
// (only the remaining parts will be retried)
Future<Integer> f = pool.take();
int index = f.get();
int index = FutureUtils.get(f);
List<Entry> batch = batches.get(index);
batches.set(index, Collections.emptyList()); // remove successful batch
// Find the most recent write time in the batch
long writeTime = batch.get(batch.size() - 1).getKey().getWriteTime();
if (writeTime > lastWriteTime) {
lastWriteTime = writeTime;
}
} catch (InterruptedException ie) {
iox = new IOException(ie);
} catch (ExecutionException ee) {
iox = ee.getCause() instanceof IOException
? (IOException) ee.getCause()
: new IOException(ee.getCause());
} catch (IOException e) {
iox = e;
} catch (RuntimeException e) {
iox = new IOException(e);
}
}
if (iox != null) {
Expand All @@ -445,7 +424,6 @@ private long parallelReplicate(CompletionService<Integer> pool, ReplicateContext
*/
@Override
public boolean replicate(ReplicateContext replicateContext) {
CompletionService<Integer> pool = new ExecutorCompletionService<>(this.exec);
int sleepMultiplier = 1;

if (!peersSelected && this.isRunning()) {
Expand All @@ -468,7 +446,7 @@ public boolean replicate(ReplicateContext replicateContext) {
}

List<List<Entry>> batches = createBatches(replicateContext.getEntries());
while (this.isRunning() && !exec.isShutdown()) {
while (this.isRunning() && !this.stopping) {
if (!isPeerEnabled()) {
if (sleepForRetries("Replication is disabled", sleepMultiplier)) {
sleepMultiplier++;
Expand All @@ -477,7 +455,7 @@ public boolean replicate(ReplicateContext replicateContext) {
}
try {
// replicate the batches to sink side.
parallelReplicate(pool, replicateContext, batches);
parallelReplicate(replicateContext, batches);
return true;
} catch (IOException ioe) {
if (ioe instanceof RemoteException) {
Expand Down Expand Up @@ -532,82 +510,117 @@ protected boolean isPeerEnabled() {

@Override
protected void doStop() {
disconnect(); // don't call super.doStop()
// Allow currently running replication tasks to finish
exec.shutdown();
try {
exec.awaitTermination(maxTerminationWait, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
}
// Abort if the tasks did not terminate in time
if (!exec.isTerminated()) {
String errMsg = "HBaseInterClusterReplicationEndpoint termination failed. The "
+ "ThreadPoolExecutor failed to finish all tasks within " + maxTerminationWait + "ms. "
+ "Aborting to prevent Replication from deadlocking. See HBASE-16081.";
abortable.abort(errMsg, new IOException(errMsg));
}
this.stopping = true;
disconnect(); // don't call super.doStop()
notifyStopped();
}

protected int replicateEntries(List<Entry> entries, int batchIndex, int timeout)
throws IOException {
protected CompletableFuture<Integer> replicateEntries(List<Entry> entries, int batchIndex,
int timeout) {
int entriesHashCode = System.identityHashCode(entries);
if (LOG.isTraceEnabled()) {
long size = entries.stream().mapToLong(this::getEstimatedEntrySize).sum();
LOG.trace("{} Replicating batch {} of {} entries with total size {} bytes to {}", logPeerId(),
entriesHashCode, entries.size(), size, replicationClusterId);
}
SinkPeer sinkPeer = null;
final CompletableFuture<Integer> resultCompletableFuture = new CompletableFuture<Integer>();
try {
int entriesHashCode = System.identityHashCode(entries);
if (LOG.isTraceEnabled()) {
long size = entries.stream().mapToLong(this::getEstimatedEntrySize).sum();
LOG.trace("{} Replicating batch {} of {} entries with total size {} bytes to {}",
logPeerId(), entriesHashCode, entries.size(), size, replicationClusterId);
}
sinkPeer = getReplicationSink();
AsyncRegionServerAdmin rsAdmin = sinkPeer.getRegionServer();
try {
ReplicationProtobufUtil.replicateWALEntry(rsAdmin,
entries.toArray(new Entry[entries.size()]), replicationClusterId, baseNamespaceDir,
hfileArchiveDir, timeout);
if (LOG.isTraceEnabled()) {
LOG.trace("{} Completed replicating batch {}", logPeerId(), entriesHashCode);
}
} catch (IOException e) {
if (LOG.isTraceEnabled()) {
LOG.trace("{} Failed replicating batch {}", logPeerId(), entriesHashCode, e);
} catch (IOException e) {
this.onReplicateWALEntryException(entriesHashCode, e, sinkPeer);
resultCompletableFuture.completeExceptionally(e);
return resultCompletableFuture;
}
assert sinkPeer != null;
AsyncRegionServerAdmin rsAdmin = sinkPeer.getRegionServer();
final SinkPeer sinkPeerToUse = sinkPeer;
FutureUtils.addListener(
ReplicationProtobufUtil.replicateWALEntry(rsAdmin, entries.toArray(new Entry[entries.size()]),
replicationClusterId, baseNamespaceDir, hfileArchiveDir, timeout),
(response, exception) -> {
if (exception != null) {
onReplicateWALEntryException(entriesHashCode, exception, sinkPeerToUse);
resultCompletableFuture.completeExceptionally(exception);
return;
}
throw e;
}
reportSinkSuccess(sinkPeer);
} catch (IOException ioe) {
reportSinkSuccess(sinkPeerToUse);
resultCompletableFuture.complete(batchIndex);
});
return resultCompletableFuture;
}

private void onReplicateWALEntryException(int entriesHashCode, Throwable exception,
final SinkPeer sinkPeer) {
if (LOG.isTraceEnabled()) {
LOG.trace("{} Failed replicating batch {}", logPeerId(), entriesHashCode, exception);
}
if (exception instanceof IOException) {
if (sinkPeer != null) {
reportBadSink(sinkPeer);
}
throw ioe;
}
return batchIndex;
}

private int serialReplicateRegionEntries(List<Entry> entries, int batchIndex, int timeout)
throws IOException {
int batchSize = 0, index = 0;
/**
* Here for {@link HBaseInterClusterReplicationEndpoint#isSerialis} is true, we iterator over the
* WAL {@link Entry} list, once we reached a batch limit, we send it out, and in the callback, we
* send the next batch, until we send all entries out.
*/
private CompletableFuture<Integer> serialReplicateRegionEntries(
PeekingIterator<Entry> walEntryPeekingIterator, int batchIndex, int timeout) {
if (!walEntryPeekingIterator.hasNext()) {
return CompletableFuture.completedFuture(batchIndex);
}
int batchSize = 0;
List<Entry> batch = new ArrayList<>();
for (Entry entry : entries) {
while (walEntryPeekingIterator.hasNext()) {
Entry entry = walEntryPeekingIterator.peek();
int entrySize = getEstimatedEntrySize(entry);
if (batchSize > 0 && batchSize + entrySize > replicationRpcLimit) {
replicateEntries(batch, index++, timeout);
batch.clear();
batchSize = 0;
break;
}
walEntryPeekingIterator.next();
batch.add(entry);
batchSize += entrySize;
}
if (batchSize > 0) {
replicateEntries(batch, index, timeout);

if (batchSize <= 0) {
return CompletableFuture.completedFuture(batchIndex);
}
return batchIndex;
final CompletableFuture<Integer> resultCompletableFuture = new CompletableFuture<Integer>();
FutureUtils.addListener(replicateEntries(batch, batchIndex, timeout), (response, exception) -> {
if (exception != null) {
resultCompletableFuture.completeExceptionally(exception);
return;
}
if (!walEntryPeekingIterator.hasNext()) {
resultCompletableFuture.complete(batchIndex);
return;
}
FutureUtils.addListener(
serialReplicateRegionEntries(walEntryPeekingIterator, batchIndex, timeout),
(currentResponse, currentException) -> {
if (currentException != null) {
resultCompletableFuture.completeExceptionally(currentException);
return;
}
resultCompletableFuture.complete(batchIndex);
});
});
return resultCompletableFuture;
}

protected Callable<Integer> createReplicator(List<Entry> entries, int batchIndex, int timeout) {
/**
* Replicate entries to peer cluster by async API.
*/
protected CompletableFuture<Integer> asyncReplicate(List<Entry> entries, int batchIndex,
int timeout) {
return isSerial
? () -> serialReplicateRegionEntries(entries, batchIndex, timeout)
: () -> replicateEntries(entries, batchIndex, timeout);
? serialReplicateRegionEntries(Iterators.peekingIterator(entries.iterator()), batchIndex,
timeout)
: replicateEntries(entries, batchIndex, timeout);
}

private String logPeerId() {
Expand Down
Loading

0 comments on commit 1e4fcac

Please sign in to comment.