comnetwork commented on code in PR #4463:
URL: https://github.com/apache/hbase/pull/4463#discussion_r889569240
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java:
##########
@@ -394,43 +374,44 @@ 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(createReplicator(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 (Throwable e) {
Review Comment:
@Apache9 , ok, I have fixed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
