You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by dl...@apache.org on 2016/03/02 22:02:54 UTC
[1/2] accumulo git commit: ACCUMULO-1755: Removed synchronization of
binning mutations in TabletServerBatchWriter
Repository: accumulo
Updated Branches:
refs/heads/1.7 85b5582b6 -> d1a9c524c
ACCUMULO-1755: Removed synchronization of binning mutations in TabletServerBatchWriter
The TabletServerBatchWriter will attempt to bin mutations in a background thread. If that
thread is busy then the binning will occur in the client thread. Previously, if binning were
to occur in one client thread, it would block all client threads from adding mutations.
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/f446b900
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/f446b900
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/f446b900
Branch: refs/heads/1.7
Commit: f446b9007ce8a4f0820e89c9e8e41a866ee8d548
Parents: b033b04
Author: Dave Marion <dl...@apache.org>
Authored: Wed Mar 2 15:08:40 2016 -0500
Committer: Dave Marion <dl...@apache.org>
Committed: Wed Mar 2 15:08:40 2016 -0500
----------------------------------------------------------------------
.../client/impl/TabletServerBatchWriter.java | 135 +++++++++++++------
.../accumulo/core/util/SimpleThreadPool.java | 6 +
.../test/functional/BatchWriterFlushIT.java | 88 +++++++++++-
3 files changed, 185 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f446b900/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java b/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
index 404b494..491bcc1 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
@@ -32,7 +32,10 @@ import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.accumulo.core.client.AccumuloException;
@@ -137,13 +140,13 @@ public class TabletServerBatchWriter {
private long initialCompileTimes;
private double initialSystemLoad;
- private int tabletServersBatchSum = 0;
- private int tabletBatchSum = 0;
- private int numBatches = 0;
- private int maxTabletBatch = Integer.MIN_VALUE;
- private int minTabletBatch = Integer.MAX_VALUE;
- private int minTabletServersBatch = Integer.MAX_VALUE;
- private int maxTabletServersBatch = Integer.MIN_VALUE;
+ private AtomicInteger tabletServersBatchSum = new AtomicInteger(0);
+ private AtomicInteger tabletBatchSum = new AtomicInteger(0);
+ private AtomicInteger numBatches = new AtomicInteger(0);
+ private AtomicInteger maxTabletBatch = new AtomicInteger(Integer.MIN_VALUE);
+ private AtomicInteger minTabletBatch = new AtomicInteger(Integer.MAX_VALUE);
+ private AtomicInteger minTabletServersBatch = new AtomicInteger(Integer.MAX_VALUE);
+ private AtomicInteger maxTabletServersBatch = new AtomicInteger(Integer.MIN_VALUE);
private Throwable lastUnknownError = null;
@@ -230,7 +233,12 @@ public class TabletServerBatchWriter {
if (mutations.getMemoryUsed() == 0)
return;
lastProcessingStartTime = System.currentTimeMillis();
- writer.addMutations(mutations);
+ try {
+ writer.queueMutations(mutations);
+ } catch (InterruptedException e) {
+ log.warn("Mutations rejected from binning thread, retrying...");
+ failedMutations.add(mutations);
+ }
mutations = new MutationSet();
}
@@ -354,6 +362,7 @@ public class TabletServerBatchWriter {
checkForFailures();
} finally {
// make a best effort to release these resources
+ writer.binningThreadPool.shutdownNow();
writer.sendThreadPool.shutdownNow();
jtimer.cancel();
span.stop();
@@ -361,26 +370,26 @@ public class TabletServerBatchWriter {
}
private void logStats() {
- long finishTime = System.currentTimeMillis();
+ if (log.isTraceEnabled()) {
+ long finishTime = System.currentTimeMillis();
- long finalGCTimes = 0;
- List<GarbageCollectorMXBean> gcmBeans = ManagementFactory.getGarbageCollectorMXBeans();
- for (GarbageCollectorMXBean garbageCollectorMXBean : gcmBeans) {
- finalGCTimes += garbageCollectorMXBean.getCollectionTime();
- }
+ long finalGCTimes = 0;
+ List<GarbageCollectorMXBean> gcmBeans = ManagementFactory.getGarbageCollectorMXBeans();
+ for (GarbageCollectorMXBean garbageCollectorMXBean : gcmBeans) {
+ finalGCTimes += garbageCollectorMXBean.getCollectionTime();
+ }
- CompilationMXBean compMxBean = ManagementFactory.getCompilationMXBean();
- long finalCompileTimes = 0;
- if (compMxBean.isCompilationTimeMonitoringSupported()) {
- finalCompileTimes = compMxBean.getTotalCompilationTime();
- }
+ CompilationMXBean compMxBean = ManagementFactory.getCompilationMXBean();
+ long finalCompileTimes = 0;
+ if (compMxBean.isCompilationTimeMonitoringSupported()) {
+ finalCompileTimes = compMxBean.getTotalCompilationTime();
+ }
- double averageRate = totalSent.get() / (totalSendTime.get() / 1000.0);
- double overallRate = totalAdded / ((finishTime - startTime) / 1000.0);
+ double averageRate = totalSent.get() / (totalSendTime.get() / 1000.0);
+ double overallRate = totalAdded / ((finishTime - startTime) / 1000.0);
- double finalSystemLoad = ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage();
+ double finalSystemLoad = ManagementFactory.getOperatingSystemMXBean().getSystemLoadAverage();
- if (log.isTraceEnabled()) {
log.trace("");
log.trace("TABLET SERVER BATCH WRITER STATISTICS");
log.trace(String.format("Added : %,10d mutations", totalAdded));
@@ -397,9 +406,10 @@ public class TabletServerBatchWriter {
log.trace(String.format("Total bin time : %,10.2f secs %6.2f%s", totalBinTime.get() / 1000.0,
100.0 * totalBinTime.get() / (finishTime - startTime), "%"));
log.trace(String.format("Average bin rate : %,10.2f mutations/sec", totalBinned.get() / (totalBinTime.get() / 1000.0)));
- log.trace(String.format("tservers per batch : %,8.2f avg %,6d min %,6d max", tabletServersBatchSum / (double) numBatches, minTabletServersBatch,
- maxTabletServersBatch));
- log.trace(String.format("tablets per batch : %,8.2f avg %,6d min %,6d max", tabletBatchSum / (double) numBatches, minTabletBatch, maxTabletBatch));
+ log.trace(String.format("tservers per batch : %,8.2f avg %,6d min %,6d max", (float) (tabletServersBatchSum.get() / numBatches.get()),
+ minTabletServersBatch.get(), maxTabletServersBatch.get()));
+ log.trace(String.format("tablets per batch : %,8.2f avg %,6d min %,6d max", (float) (tabletBatchSum.get() / numBatches.get()), minTabletBatch.get(),
+ maxTabletBatch.get()));
log.trace("");
log.trace("SYSTEM STATISTICS");
log.trace(String.format("JVM GC Time : %,10.2f secs", ((finalGCTimes - initialGCTimes) / 1000.0)));
@@ -416,16 +426,32 @@ public class TabletServerBatchWriter {
}
public void updateBinningStats(int count, long time, Map<String,TabletServerMutations<Mutation>> binnedMutations) {
- totalBinTime.addAndGet(time);
- totalBinned.addAndGet(count);
- updateBatchStats(binnedMutations);
+ if (log.isTraceEnabled()) {
+ totalBinTime.addAndGet(time);
+ totalBinned.addAndGet(count);
+ updateBatchStats(binnedMutations);
+ }
}
- private synchronized void updateBatchStats(Map<String,TabletServerMutations<Mutation>> binnedMutations) {
- tabletServersBatchSum += binnedMutations.size();
+ private static void computeMin(AtomicInteger stat, int update) {
+ int old = stat.get();
+ while (!stat.compareAndSet(old, Math.min(old, update))) {
+ old = stat.get();
+ }
+ }
+
+ private static void computeMax(AtomicInteger stat, int update) {
+ int old = stat.get();
+ while (!stat.compareAndSet(old, Math.max(old, update))) {
+ old = stat.get();
+ }
+ }
- minTabletServersBatch = Math.min(minTabletServersBatch, binnedMutations.size());
- maxTabletServersBatch = Math.max(maxTabletServersBatch, binnedMutations.size());
+ private void updateBatchStats(Map<String,TabletServerMutations<Mutation>> binnedMutations) {
+ tabletServersBatchSum.addAndGet(binnedMutations.size());
+
+ computeMin(minTabletServersBatch, binnedMutations.size());
+ computeMax(maxTabletServersBatch, binnedMutations.size());
int numTablets = 0;
@@ -434,12 +460,12 @@ public class TabletServerBatchWriter {
numTablets += tsm.getMutations().size();
}
- tabletBatchSum += numTablets;
+ tabletBatchSum.addAndGet(numTablets);
- minTabletBatch = Math.min(minTabletBatch, numTablets);
- maxTabletBatch = Math.max(maxTabletBatch, numTablets);
+ computeMin(minTabletBatch, numTablets);
+ computeMax(maxTabletBatch, numTablets);
- numBatches++;
+ numBatches.incrementAndGet();
}
private void waitRTE() {
@@ -616,19 +642,22 @@ public class TabletServerBatchWriter {
private class MutationWriter {
private static final int MUTATION_BATCH_SIZE = 1 << 17;
- private ExecutorService sendThreadPool;
- private Map<String,TabletServerMutations<Mutation>> serversMutations;
- private Set<String> queued;
- private Map<String,TabletLocator> locators;
+ private final ExecutorService sendThreadPool;
+ private final SimpleThreadPool binningThreadPool;
+ private final Map<String,TabletServerMutations<Mutation>> serversMutations;
+ private final Set<String> queued;
+ private final Map<String,TabletLocator> locators;
public MutationWriter(int numSendThreads) {
serversMutations = new HashMap<String,TabletServerMutations<Mutation>>();
queued = new HashSet<String>();
sendThreadPool = new SimpleThreadPool(numSendThreads, this.getClass().getName());
locators = new HashMap<String,TabletLocator>();
+ binningThreadPool = new SimpleThreadPool(1, "BinMutations", new SynchronousQueue<Runnable>());
+ binningThreadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
}
- private TabletLocator getLocator(String tableId) {
+ private synchronized TabletLocator getLocator(String tableId) {
TabletLocator ret = locators.get(tableId);
if (ret == null) {
ret = TabletLocator.getLocator(instance, new Text(tableId));
@@ -686,7 +715,27 @@ public class TabletServerBatchWriter {
}
- void addMutations(MutationSet mutationsToSend) {
+ void queueMutations(final MutationSet mutationsToSend) throws InterruptedException {
+ if (null == mutationsToSend)
+ return;
+ binningThreadPool.execute(new Runnable() {
+
+ @Override
+ public void run() {
+ if (null != mutationsToSend) {
+ try {
+ if (log.isTraceEnabled())
+ log.trace(Thread.currentThread().getName() + " - binning " + mutationsToSend.size() + " mutations");
+ addMutations(mutationsToSend);
+ } catch (Exception e) {
+ updateUnknownErrors("Error processing mutation set", e);
+ }
+ }
+ }
+ });
+ }
+
+ private void addMutations(MutationSet mutationsToSend) {
Map<String,TabletServerMutations<Mutation>> binnedMutations = new HashMap<String,TabletServerMutations<Mutation>>();
Span span = Trace.start("binMutations");
try {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f446b900/core/src/main/java/org/apache/accumulo/core/util/SimpleThreadPool.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/util/SimpleThreadPool.java b/core/src/main/java/org/apache/accumulo/core/util/SimpleThreadPool.java
index a406233..8991991 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/SimpleThreadPool.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/SimpleThreadPool.java
@@ -16,6 +16,7 @@
*/
package org.apache.accumulo.core.util;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -30,4 +31,9 @@ public class SimpleThreadPool extends ThreadPoolExecutor {
allowCoreThreadTimeOut(true);
}
+ public SimpleThreadPool(int max, final String name, BlockingQueue<Runnable> queue) {
+ super(max, max, 4l, TimeUnit.SECONDS, queue, new NamingThreadFactory(name));
+ allowCoreThreadTimeOut(true);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/f446b900/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java b/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
index 52d9c93..e2277a3 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
@@ -18,9 +18,16 @@ package org.apache.accumulo.test.functional;
import static com.google.common.base.Charsets.UTF_8;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Map.Entry;
import java.util.Random;
+import java.util.Set;
+import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.client.AccumuloException;
@@ -36,14 +43,17 @@ import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.SimpleThreadPool;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.harness.AccumuloClusterIT;
import org.apache.hadoop.io.Text;
+import org.junit.Assert;
import org.junit.Test;
public class BatchWriterFlushIT extends AccumuloClusterIT {
private static final int NUM_TO_FLUSH = 100000;
+ private static final int NUM_THREADS = 3;
@Override
protected int defaultTimeoutSeconds() {
@@ -60,7 +70,6 @@ public class BatchWriterFlushIT extends AccumuloClusterIT {
c.tableOperations().create(bwlt);
runFlushTest(bwft);
runLatencyTest(bwlt);
-
}
private void runLatencyTest(String tableName) throws Exception {
@@ -170,6 +179,83 @@ public class BatchWriterFlushIT extends AccumuloClusterIT {
}
}
+ @Test
+ public void runMultiThreadedBinningTest() throws Exception {
+ Connector c = getConnector();
+ String[] tableNames = getUniqueNames(1);
+ String tableName = tableNames[0];
+ c.tableOperations().create(tableName);
+ for (int x = 0; x < NUM_THREADS; x++) {
+ c.tableOperations().addSplits(tableName, new TreeSet<Text>(Collections.singleton(new Text(Integer.toString(x * NUM_TO_FLUSH)))));
+ }
+
+ // Logger.getLogger(TabletServerBatchWriter.class).setLevel(Level.TRACE);
+ final List<Set<Mutation>> allMuts = new LinkedList<Set<Mutation>>();
+ List<Mutation> data = new ArrayList<Mutation>();
+ for (int i = 0; i < NUM_THREADS; i++) {
+ final int thread = i;
+ for (int j = 0; j < NUM_TO_FLUSH; j++) {
+ int row = thread * NUM_TO_FLUSH + j;
+ Mutation m = new Mutation(new Text(String.format("%10d", row)));
+ m.put(new Text("cf" + thread), new Text("cq"), new Value(("" + row).getBytes()));
+ data.add(m);
+ }
+ }
+ Assert.assertEquals(NUM_THREADS * NUM_TO_FLUSH, data.size());
+ Collections.shuffle(data);
+ for (int n = 0; n < (NUM_THREADS * NUM_TO_FLUSH); n += NUM_TO_FLUSH) {
+ Set<Mutation> muts = new HashSet<Mutation>(data.subList(n, n + NUM_TO_FLUSH));
+ allMuts.add(muts);
+ }
+
+ SimpleThreadPool threads = new SimpleThreadPool(NUM_THREADS, "ClientThreads");
+ threads.allowCoreThreadTimeOut(false);
+ threads.prestartAllCoreThreads();
+
+ BatchWriterConfig cfg = new BatchWriterConfig();
+ cfg.setMaxLatency(10, TimeUnit.SECONDS);
+ cfg.setMaxMemory(1 * 1024 * 1024);
+ cfg.setMaxWriteThreads(NUM_THREADS);
+ final BatchWriter bw = getConnector().createBatchWriter(tableName, cfg);
+
+ for (int k = 0; k < NUM_THREADS; k++) {
+ final int idx = k;
+ threads.execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ bw.addMutations(allMuts.get(idx));
+ bw.flush();
+ } catch (MutationsRejectedException e) {
+ Assert.fail("Error adding mutations to batch writer");
+ }
+ }
+ });
+ }
+ threads.shutdown();
+ threads.awaitTermination(3, TimeUnit.MINUTES);
+ bw.close();
+ Scanner scanner = getConnector().createScanner(tableName, Authorizations.EMPTY);
+ for (Entry<Key,Value> e : scanner) {
+ Mutation m = new Mutation(e.getKey().getRow());
+ m.put(e.getKey().getColumnFamily(), e.getKey().getColumnQualifier(), e.getValue());
+ boolean found = false;
+ for (int l = 0; l < NUM_THREADS; l++) {
+ if (allMuts.get(l).contains(m)) {
+ found = true;
+ allMuts.get(l).remove(m);
+ break;
+ }
+ }
+ Assert.assertTrue("Mutation not found: " + m.toString(), found);
+ }
+
+ for (int m = 0; m < NUM_THREADS; m++) {
+ Assert.assertEquals(0, allMuts.get(m).size());
+ }
+
+ }
+
private void verifyEntry(int row, Entry<Key,Value> entry) throws Exception {
if (!entry.getKey().getRow().toString().equals(String.format("r_%10d", row))) {
throw new Exception("Unexpected key returned, expected " + row + " got " + entry.getKey());
[2/2] accumulo git commit: Merge branch '1.6' into 1.7
Posted by dl...@apache.org.
Merge branch '1.6' into 1.7
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/d1a9c524
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/d1a9c524
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/d1a9c524
Branch: refs/heads/1.7
Commit: d1a9c524c69b9ea650d9579ed7f3272fa8d2a8e5
Parents: 85b5582 f446b90
Author: Dave Marion <dl...@apache.org>
Authored: Wed Mar 2 15:38:20 2016 -0500
Committer: Dave Marion <dl...@apache.org>
Committed: Wed Mar 2 15:38:20 2016 -0500
----------------------------------------------------------------------
.../client/impl/TabletServerBatchWriter.java | 126 +++++++++++++------
.../accumulo/core/util/SimpleThreadPool.java | 6 +
.../test/functional/BatchWriterFlushIT.java | 88 ++++++++++++-
3 files changed, 180 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d1a9c524/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
----------------------------------------------------------------------
diff --cc core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
index f14116a,491bcc1..8922ac5
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
@@@ -136,27 -140,22 +139,27 @@@ public class TabletServerBatchWriter
private long initialCompileTimes;
private double initialSystemLoad;
- private int tabletServersBatchSum = 0;
- private int tabletBatchSum = 0;
- private int numBatches = 0;
- private int maxTabletBatch = Integer.MIN_VALUE;
- private int minTabletBatch = Integer.MAX_VALUE;
- private int minTabletServersBatch = Integer.MAX_VALUE;
- private int maxTabletServersBatch = Integer.MIN_VALUE;
+ private AtomicInteger tabletServersBatchSum = new AtomicInteger(0);
+ private AtomicInteger tabletBatchSum = new AtomicInteger(0);
+ private AtomicInteger numBatches = new AtomicInteger(0);
+ private AtomicInteger maxTabletBatch = new AtomicInteger(Integer.MIN_VALUE);
+ private AtomicInteger minTabletBatch = new AtomicInteger(Integer.MAX_VALUE);
+ private AtomicInteger minTabletServersBatch = new AtomicInteger(Integer.MAX_VALUE);
+ private AtomicInteger maxTabletServersBatch = new AtomicInteger(Integer.MIN_VALUE);
+ // error handling
+ private final Violations violations = new Violations();
+ private final Map<KeyExtent,Set<SecurityErrorCode>> authorizationFailures = new HashMap<KeyExtent,Set<SecurityErrorCode>>();
+ private final HashSet<String> serverSideErrors = new HashSet<String>();
+ private final FailedMutations failedMutations = new FailedMutations();
+ private int unknownErrors = 0;
+ private boolean somethingFailed = false;
private Throwable lastUnknownError = null;
- private Map<String,TimeoutTracker> timeoutTrackers;
-
private static class TimeoutTracker {
- String server;
- long timeOut;
+ final String server;
+ final long timeOut;
long activityTime;
Long firstErrorTime = null;
@@@ -441,23 -460,17 +467,23 @@@
numTablets += tsm.getMutations().size();
}
- tabletBatchSum += numTablets;
+ tabletBatchSum.addAndGet(numTablets);
- minTabletBatch = Math.min(minTabletBatch, numTablets);
- maxTabletBatch = Math.max(maxTabletBatch, numTablets);
+ computeMin(minTabletBatch, numTablets);
+ computeMax(maxTabletBatch, numTablets);
- numBatches++;
+ numBatches.incrementAndGet();
}
- private void waitRTE() {
+ private interface WaitCondition {
+ boolean shouldWait();
+ }
+
+ private void waitRTE(WaitCondition condition) {
try {
- wait();
+ while (condition.shouldWait()) {
+ wait();
+ }
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
@@@ -639,12 -653,14 +666,14 @@@
queued = new HashSet<String>();
sendThreadPool = new SimpleThreadPool(numSendThreads, this.getClass().getName());
locators = new HashMap<String,TabletLocator>();
+ binningThreadPool = new SimpleThreadPool(1, "BinMutations", new SynchronousQueue<Runnable>());
+ binningThreadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
}
- private TabletLocator getLocator(String tableId) {
+ private synchronized TabletLocator getLocator(String tableId) {
TabletLocator ret = locators.get(tableId);
if (ret == null) {
- ret = TabletLocator.getLocator(instance, new Text(tableId));
+ ret = TabletLocator.getLocator(context, new Text(tableId));
ret = new TimeoutTabletLocator(ret, timeout);
locators.put(tableId, ret);
}
@@@ -701,7 -715,27 +730,26 @@@
}
- void addMutations(MutationSet mutationsToSend) {
+ void queueMutations(final MutationSet mutationsToSend) throws InterruptedException {
+ if (null == mutationsToSend)
+ return;
+ binningThreadPool.execute(new Runnable() {
+
+ @Override
+ public void run() {
+ if (null != mutationsToSend) {
+ try {
- if (log.isTraceEnabled())
- log.trace(Thread.currentThread().getName() + " - binning " + mutationsToSend.size() + " mutations");
++ log.trace("{} - binning {} mutations", Thread.currentThread().getName(), mutationsToSend.size());
+ addMutations(mutationsToSend);
+ } catch (Exception e) {
+ updateUnknownErrors("Error processing mutation set", e);
+ }
+ }
+ }
+ });
+ }
+
+ private void addMutations(MutationSet mutationsToSend) {
Map<String,TabletServerMutations<Mutation>> binnedMutations = new HashMap<String,TabletServerMutations<Mutation>>();
Span span = Trace.start("binMutations");
try {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/d1a9c524/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
----------------------------------------------------------------------
diff --cc test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
index 6378480,e2277a3..63bee16
--- a/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/BatchWriterFlushIT.java
@@@ -16,11 -16,18 +16,18 @@@
*/
package org.apache.accumulo.test.functional;
-import static com.google.common.base.Charsets.UTF_8;
+import static java.nio.charset.StandardCharsets.UTF_8;
+ import java.util.ArrayList;
+ import java.util.Collections;
+ import java.util.HashSet;
import java.util.Iterator;
+ import java.util.LinkedList;
+ import java.util.List;
import java.util.Map.Entry;
import java.util.Random;
+ import java.util.Set;
+ import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.client.AccumuloException;
@@@ -39,10 -47,9 +47,11 @@@ import org.apache.accumulo.core.util.Si
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.harness.AccumuloClusterIT;
import org.apache.hadoop.io.Text;
+ import org.junit.Assert;
import org.junit.Test;
+import com.google.common.collect.Iterators;
+
public class BatchWriterFlushIT extends AccumuloClusterIT {
private static final int NUM_TO_FLUSH = 100000;